diff --git a/backend/internal/flow/flowexec/model.go b/backend/internal/flow/flowexec/model.go index ded0d4d740..51c646fbea 100644 --- a/backend/internal/flow/flowexec/model.go +++ b/backend/internal/flow/flowexec/model.go @@ -90,9 +90,9 @@ type EngineContext struct { // SSOHandleIn carries the inbound SSO handle for this request. It is transient: read from // the transport at the start of execution and never persisted with the flow context. SSOHandleIn string - // SSOFlowVersion is the current active version of this flow's definition, resolved via - // the flow management provider. Transient; used by the SSO-Check node to reject sessions - // established at an incompatible flow version. + // SSOFlowVersion is the current active version of this flow's definition, captured from the + // flow fetched when the context is loaded. Transient; used by the SSO-Check node to reject + // sessions established at an incompatible flow version. SSOFlowVersion int } diff --git a/backend/internal/flow/flowexec/service.go b/backend/internal/flow/flowexec/service.go index 9d2960183e..3ee1ffceb2 100644 --- a/backend/internal/flow/flowexec/service.go +++ b/backend/internal/flow/flowexec/service.go @@ -32,7 +32,6 @@ import ( "github.com/thunder-id/thunderid/internal/flow/common" flowconfig "github.com/thunder-id/thunderid/internal/flow/config" "github.com/thunder-id/thunderid/internal/flow/core" - "github.com/thunder-id/thunderid/internal/flow/executor" "github.com/thunder-id/thunderid/internal/flow/graphbuilder" "github.com/thunder-id/thunderid/internal/flow/session" sysContext "github.com/thunder-id/thunderid/internal/system/context" @@ -132,14 +131,6 @@ func (s *flowExecService) Execute(ctx context.Context, // Resolve the inbound SSO handle for this flow from the request-scoped transport inputs. applyInboundSSO(engineCtx, ctx) - // Resolve the active flow version whenever the flow establishes or consults an SSO session. - // Both paths need it: the save path (fresh login, which carries no inbound handle) stamps the - // version onto the new session, and the check path compares against it. Gating this on an - // inbound handle would save sessions at version 0 and then fail the version check on the next - // login. Flows that use no SSO session skip the lookup. - if flowUsesSSOSession(engineCtx.Graph) { - engineCtx.SSOFlowVersion = s.resolveActiveFlowVersion(ctx, engineCtx, logger) - } flowStep, flowErr := s.flowEngine.Execute(engineCtx) @@ -195,43 +186,6 @@ func applyInboundSSO(engineCtx *EngineContext, ctx context.Context) { engineCtx.SSOHandleIn = inbound.HandleFor(engineCtx.Graph.GetID()) } -// flowUsesSSOSession reports whether the flow graph contains a node that establishes or -// consults an SSO session (the Session or SSO-Check executors). It gates the active-flow-version -// lookup so only SSO-capable flows pay for it. -func flowUsesSSOSession(graph core.GraphInterface) bool { - if graph == nil { - return false - } - for _, node := range graph.GetNodes() { - execNode, ok := node.(core.ExecutorBackedNodeInterface) - if !ok { - continue - } - name := execNode.GetExecutorName() - if name == executor.ExecutorNameSession || name == executor.ExecutorNameSSOCheck { - return true - } - } - return false -} - -// resolveActiveFlowVersion returns the current active version of the flow, or 0 when it -// cannot be determined. A 0 (unknown) version makes the SSO-Check node treat any saved -// session as version-incompatible, i.e. it falls back to full authentication. -func (s *flowExecService) resolveActiveFlowVersion(ctx context.Context, engineCtx *EngineContext, - logger *log.Logger) int { - if engineCtx.Graph == nil { - return 0 - } - def, svcErr := s.flowProvider.GetFlow(ctx, engineCtx.Graph.GetID()) - if svcErr != nil || def == nil { - logger.Debug(ctx, "Could not resolve active flow version for SSO check", - log.String("flowID", engineCtx.Graph.GetID())) - return 0 - } - return def.ActiveVersion -} - // initContext initializes a new flow context with the given details. func (s *flowExecService) loadNewContext(ctx context.Context, appID, flowTypeStr string, verbose bool, action string, inputs map[string]string, flowSecret string, logger *log.Logger) ( @@ -368,6 +322,7 @@ func (s *flowExecService) initContext(ctx context.Context, appID string, flowTyp } engineCtx.FlowType = flow.FlowType + engineCtx.SSOFlowVersion = flow.ActiveVersion graph, svcErr := s.graphBuilder.GetGraph(ctx, flow) if svcErr != nil { logger.Error(ctx, "Error retrieving graph from graph builder", @@ -467,6 +422,7 @@ func (s *flowExecService) loadContextFromStore(ctx context.Context, executionID log.String(log.LoggerKeyExecutionID, executionID), log.Error(err)) return nil, &tidcommon.InternalServerError } + engineContext.SSOFlowVersion = flow.ActiveVersion // Set application context if required if err := s.setApplicationToContext(&engineContext, logger); err != nil { diff --git a/backend/internal/flow/flowexec/service_sso_test.go b/backend/internal/flow/flowexec/service_sso_test.go index c3c28bb164..0183a97d84 100644 --- a/backend/internal/flow/flowexec/service_sso_test.go +++ b/backend/internal/flow/flowexec/service_sso_test.go @@ -22,16 +22,12 @@ import ( "context" "testing" - "github.com/stretchr/testify/mock" "github.com/stretchr/testify/suite" - "github.com/thunder-id/thunderid/internal/flow/common" "github.com/thunder-id/thunderid/internal/flow/core" - "github.com/thunder-id/thunderid/internal/flow/executor" "github.com/thunder-id/thunderid/internal/flow/session" "github.com/thunder-id/thunderid/internal/system/cache" "github.com/thunder-id/thunderid/internal/system/config" - "github.com/thunder-id/thunderid/internal/system/log" "github.com/thunder-id/thunderid/pkg/thunderidengine/providers" ) @@ -58,17 +54,6 @@ func (s *ServiceSSOTestSuite) newTestGraph() core.GraphInterface { return flowFactory.CreateGraph(testFlowID, providers.FlowTypeAuthentication, 1) } -// newGraphWithExecutor builds a single-node graph whose node is backed by the given executor. -func (s *ServiceSSOTestSuite) newGraphWithExecutor(executorName string) core.GraphInterface { - flowFactory, _ := core.Initialize(cache.Initialize(config.GetServerRuntime().Config.Cache, "test-deployment")) - graph := flowFactory.CreateGraph(testFlowID, providers.FlowTypeAuthentication, 1) - node, err := flowFactory.CreateNode("n1", string(common.NodeTypeTaskExecution), nil, false, false) - s.Require().NoError(err) - node.(core.ExecutorBackedNodeInterface).SetExecutorName(executorName) - s.Require().NoError(graph.AddNode(node)) - return graph -} - func (s *ServiceSSOTestSuite) TestApplyInboundSSO_SelectsHandleForFlow() { engineCtx := &EngineContext{Graph: s.newTestGraph()} @@ -99,35 +84,3 @@ func (s *ServiceSSOTestSuite) TestApplyInboundSSO_NilGraph() { s.Empty(engineCtx.SSOHandleIn) } - -func (s *ServiceSSOTestSuite) TestResolveActiveFlowVersion_FromProvider() { - provider := NewFlowProviderMock(s.T()) - provider.EXPECT().GetFlow(mock.Anything, testFlowID). - Return(&providers.CompleteFlowDefinition{ID: testFlowID, ActiveVersion: 5}, nil) - svc := &flowExecService{flowProvider: provider} - engineCtx := &EngineContext{Graph: s.newTestGraph()} - - version := svc.resolveActiveFlowVersion(context.Background(), engineCtx, log.GetLogger()) - - s.Equal(5, version) -} - -func (s *ServiceSSOTestSuite) TestResolveActiveFlowVersion_NilGraph() { - // A nil graph short-circuits before the provider is consulted, so no GetFlow call is expected. - svc := &flowExecService{flowProvider: NewFlowProviderMock(s.T())} - - version := svc.resolveActiveFlowVersion(context.Background(), &EngineContext{}, log.GetLogger()) - - s.Equal(0, version) -} - -// TestFlowUsesSSOSession covers the version-lookup gate: a flow that establishes (Session) or consults -// (SSO-Check) a session must have its active version resolved on every path — including the -// fresh-login save path, which carries no inbound handle. Gating the version lookup on an inbound -// handle would persist sessions at version 0 and fail the version check on the next login. -func (s *ServiceSSOTestSuite) TestFlowUsesSSOSession() { - s.True(flowUsesSSOSession(s.newGraphWithExecutor(executor.ExecutorNameSession))) - s.True(flowUsesSSOSession(s.newGraphWithExecutor(executor.ExecutorNameSSOCheck))) - s.False(flowUsesSSOSession(s.newGraphWithExecutor(executor.ExecutorNameCredentialsAuth))) - s.False(flowUsesSSOSession(nil)) -} diff --git a/backend/internal/flow/graphbuilder/graph_builder_test.go b/backend/internal/flow/graphbuilder/graph_builder_test.go index 70cc778f92..dd1a53e01e 100644 --- a/backend/internal/flow/graphbuilder/graph_builder_test.go +++ b/backend/internal/flow/graphbuilder/graph_builder_test.go @@ -20,9 +20,7 @@ package graphbuilder import ( "context" - "encoding/json" "errors" - "os" "testing" "github.com/thunder-id/thunderid/pkg/thunderidengine/providers" @@ -1965,12 +1963,62 @@ func (s *GraphBuilderTestSuite) TestConfigureNodePrompts_InvalidRegexFailsBuild( // registered executors, and builds into a graph with all expected nodes. It exercises the // real executor registry so a typo'd executor name or dangling node reference fails here. func (s *GraphBuilderTestSuite) TestSSOFlowDefinitionBuilds() { - raw, err := os.ReadFile("testdata/sso_flow.json") - s.Require().NoError(err) - - var def providers.CompleteFlowDefinition - s.Require().NoError(json.Unmarshal(raw, &def)) - def.ID = "auth-sso-flow-test" + def := providers.CompleteFlowDefinition{ + ID: "auth-sso-flow-test", + Name: "Default SSO Authentication Flow", + Handle: "default-sso-flow", + FlowType: providers.FlowTypeAuthentication, + Nodes: []providers.NodeDefinition{ + {ID: "start", Type: "START", OnSuccess: "sso_check"}, + { + ID: "sso_check", + Type: "TASK_EXECUTION", + Executor: &providers.ExecutorDefinition{Name: executor.ExecutorNameSSOCheck}, + Properties: map[string]interface{}{common.NodePropertyCheckpointRef: "session"}, + OnSuccess: "session", + OnFailure: "prompt_credentials", + }, + { + ID: "basic_auth", + Type: "TASK_EXECUTION", + Executor: &providers.ExecutorDefinition{Name: executor.ExecutorNameCredentialsAuth}, + OnSuccess: "session", + OnIncomplete: "prompt_credentials", + }, + { + ID: "prompt_credentials", + Type: "PROMPT", + Prompts: []providers.PromptDefinition{ + { + Inputs: []providers.InputDefinition{ + {Ref: "input_001", Identifier: "username", Type: "TEXT_INPUT", Required: true}, + {Ref: "input_002", Identifier: "password", Type: "PASSWORD_INPUT", Required: true}, + }, + Action: &providers.ActionDefinition{Ref: "action_001", NextNode: "basic_auth"}, + }, + }, + }, + { + ID: "session", + Type: "TASK_EXECUTION", + Executor: &providers.ExecutorDefinition{Name: executor.ExecutorNameSession}, + OnSuccess: "authorization_check", + }, + { + ID: "authorization_check", + Type: "TASK_EXECUTION", + Executor: &providers.ExecutorDefinition{Name: executor.ExecutorNameAuthorization}, + OnSuccess: "auth_assert", + }, + { + ID: "auth_assert", + Type: "TASK_EXECUTION", + Executor: &providers.ExecutorDefinition{Name: executor.ExecutorNameAuthAssert}, + OnSuccess: "end", + }, + {ID: "end", Type: "END"}, + }, + } flowFactory, graphCache := core.Initialize( cache.Initialize(config.GetServerRuntime().Config.Cache, "test-deployment")) diff --git a/backend/internal/flow/graphbuilder/testdata/sso_flow.json b/backend/internal/flow/graphbuilder/testdata/sso_flow.json deleted file mode 100644 index 3b40cbcf67..0000000000 --- a/backend/internal/flow/graphbuilder/testdata/sso_flow.json +++ /dev/null @@ -1,127 +0,0 @@ -{ - "name": "Default SSO Authentication Flow", - "handle": "default-sso-flow", - "flowType": "AUTHENTICATION", - "nodes": [ - { - "id": "start", - "type": "START", - "onSuccess": "sso_check" - }, - { - "id": "sso_check", - "type": "TASK_EXECUTION", - "executor": { - "name": "SSOCheckExecutor" - }, - "properties": { - "checkpointRef": "session" - }, - "onSuccess": "session", - "onFailure": "prompt_credentials" - }, - { - "id": "basic_auth", - "type": "TASK_EXECUTION", - "executor": { - "name": "CredentialsAuthExecutor" - }, - "onSuccess": "session", - "onIncomplete": "prompt_credentials" - }, - { - "id": "prompt_credentials", - "type": "PROMPT", - "meta": { - "components": [ - { - "align": "center", - "type": "TEXT", - "id": "text_001", - "label": "{{ t(signin:forms.credentials.title) }}", - "variant": "HEADING_1" - }, - { - "type": "BLOCK", - "id": "block_001", - "components": [ - { - "id": "input_001", - "ref": "username", - "type": "TEXT_INPUT", - "label": "{{ t(signin:forms.credentials.fields.username.label) }}", - "required": true, - "placeholder": "{{ t(signin:forms.credentials.fields.username.placeholder) }}" - }, - { - "id": "input_002", - "ref": "password", - "type": "PASSWORD_INPUT", - "label": "{{ t(signin:forms.credentials.fields.password.label) }}", - "required": true, - "placeholder": "{{ t(signin:forms.credentials.fields.password.placeholder) }}" - }, - { - "type": "ACTION", - "id": "action_001", - "label": "{{ t(signin:forms.credentials.actions.submit.label) }}", - "variant": "PRIMARY", - "eventType": "SUBMIT" - } - ] - } - ] - }, - "prompts": [ - { - "inputs": [ - { - "ref": "input_001", - "identifier": "username", - "type": "TEXT_INPUT", - "required": true - }, - { - "ref": "input_002", - "identifier": "password", - "type": "PASSWORD_INPUT", - "required": true - } - ], - "action": { - "ref": "action_001", - "nextNode": "basic_auth" - } - } - ] - }, - { - "id": "session", - "type": "TASK_EXECUTION", - "executor": { - "name": "SessionExecutor" - }, - "onSuccess": "authorization_check" - }, - { - "id": "authorization_check", - "type": "TASK_EXECUTION", - "executor": { - "name": "AuthorizationExecutor" - }, - "onSuccess": "auth_assert" - }, - { - "id": "auth_assert", - "type": "TASK_EXECUTION", - "executor": { - "name": "AuthAssertExecutor" - }, - "onSuccess": "end" - }, - { - "id": "end", - "type": "END" - } - ] -} diff --git a/backend/internal/flow/session/participant_store.go b/backend/internal/flow/session/participant_store.go deleted file mode 100644 index 1cbbadef93..0000000000 --- a/backend/internal/flow/session/participant_store.go +++ /dev/null @@ -1,100 +0,0 @@ -/* - * Copyright (c) 2026, WSO2 LLC. (https://www.wso2.com). - * - * WSO2 LLC. licenses this file to you under the Apache License, - * Version 2.0 (the "License"); you may not use this file except - * in compliance with the License. - * You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, - * software distributed under the License is distributed on an - * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY - * KIND, either express or implied. See the License for the - * specific language governing permissions and limitations - * under the License. - */ - -package session - -import ( - "context" - "fmt" - - "github.com/thunder-id/thunderid/internal/system/database/provider" - sysutils "github.com/thunder-id/thunderid/internal/system/utils" -) - -// Record inserts or refreshes a participant under the upsert query. -func (st *store) Record(ctx context.Context, p Participant) error { - return withOperationDBClient(st.dbProvider, func(dbClient provider.DBClientInterface) error { - _, err := dbClient.ExecuteContext(ctx, queryUpsertParticipant, - p.SessionID, st.deploymentID, p.AppID, p.FirstJoinedAt, p.LastActiveAt) - if err != nil { - return fmt.Errorf("failed to record session participant: %w", err) - } - return nil - }) -} - -// ListBySessionID returns the participants of a session, oldest first. -func (st *store) ListBySessionID(ctx context.Context, sessionID string) ([]Participant, error) { - var result []Participant - - err := withOperationDBClient(st.dbProvider, func(dbClient provider.DBClientInterface) error { - results, queryErr := dbClient.QueryContext(ctx, queryListParticipantsBySessionID, sessionID, st.deploymentID) - if queryErr != nil { - return fmt.Errorf("failed to execute query: %w", queryErr) - } - for _, row := range results { - p, buildErr := buildParticipantFromRow(row) - if buildErr != nil { - return buildErr - } - result = append(result, p) - } - return nil - }) - if err != nil { - return nil, err - } - return result, nil -} - -// DeleteBySessionID removes all participants of a session. -func (st *store) DeleteBySessionID(ctx context.Context, sessionID string) error { - return withOperationDBClient(st.dbProvider, func(dbClient provider.DBClientInterface) error { - _, err := dbClient.ExecuteContext(ctx, queryDeleteParticipantsBySessionID, sessionID, st.deploymentID) - if err != nil { - return fmt.Errorf("failed to delete session participants: %w", err) - } - return nil - }) -} - -// buildParticipantFromRow maps a database result row into a Participant. -func buildParticipantFromRow(row map[string]interface{}) (Participant, error) { - sessionID, err := parseString(row["session_id"], "session_id") - if err != nil { - return Participant{}, err - } - appID, err := parseString(row["app_id"], "app_id") - if err != nil { - return Participant{}, err - } - firstJoinedAt, err := sysutils.ParseDBTimeField(row["first_joined_at"], "first_joined_at") - if err != nil { - return Participant{}, err - } - lastActiveAt, err := sysutils.ParseDBTimeField(row["last_active_at"], "last_active_at") - if err != nil { - return Participant{}, err - } - return Participant{ - SessionID: sessionID, - AppID: appID, - FirstJoinedAt: firstJoinedAt, - LastActiveAt: lastActiveAt, - }, nil -} diff --git a/backend/internal/flow/session/session_context_store.go b/backend/internal/flow/session/session_context_store.go deleted file mode 100644 index 690e80d22d..0000000000 --- a/backend/internal/flow/session/session_context_store.go +++ /dev/null @@ -1,141 +0,0 @@ -/* - * Copyright (c) 2026, WSO2 LLC. (https://www.wso2.com). - * - * WSO2 LLC. licenses this file to you under the Apache License, - * Version 2.0 (the "License"); you may not use this file except - * in compliance with the License. - * You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, - * software distributed under the License is distributed on an - * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY - * KIND, either express or implied. See the License for the - * specific language governing permissions and limitations - * under the License. - */ - -package session - -import ( - "context" - "fmt" - - "github.com/thunder-id/thunderid/internal/system/database/provider" -) - -// CreateContext persists the session context. It rejects payloads exceeding MaxSessionContextBytes. -func (st *store) CreateContext(ctx context.Context, c SessionContext) error { - payload, err := c.serializePayload() - if err != nil { - return err - } - if len(payload) > MaxSessionContextBytes { - return errSessionContextTooLarge - } - - return withOperationDBClient(st.dbProvider, func(dbClient provider.DBClientInterface) error { - _, execErr := dbClient.ExecuteContext(ctx, queryCreateSessionContext, - c.SessionID, st.deploymentID, c.CheckpointID, payload, c.ContextVersion) - if execErr != nil { - return fmt.Errorf("failed to create session context: %w", execErr) - } - return nil - }) -} - -// GetByCheckpoint fetches one checkpoint's session context. -func (st *store) GetByCheckpoint(ctx context.Context, sessionID, - checkpointID string) (*SessionContext, error) { - var result *SessionContext - - err := withOperationDBClient(st.dbProvider, func(dbClient provider.DBClientInterface) error { - results, queryErr := dbClient.QueryContext(ctx, queryGetSessionContextByCheckpoint, - sessionID, st.deploymentID, checkpointID) - if queryErr != nil { - return fmt.Errorf("failed to execute query: %w", queryErr) - } - if len(results) == 0 { - return nil - } - if len(results) != 1 { - return fmt.Errorf("unexpected number of results: %d", len(results)) - } - - c, buildErr := st.buildSessionContextFromRow(results[0]) - if buildErr != nil { - return buildErr - } - result = c - return nil - }) - if err != nil { - return nil, err - } - return result, nil -} - -// ListCheckpointIDs returns the checkpoint ids a session has saved, without decrypting any payload. -func (st *store) ListCheckpointIDs(ctx context.Context, sessionID string) ([]string, error) { - var ids []string - - err := withOperationDBClient(st.dbProvider, func(dbClient provider.DBClientInterface) error { - results, queryErr := dbClient.QueryContext(ctx, queryListCheckpointsBySessionID, sessionID, st.deploymentID) - if queryErr != nil { - return fmt.Errorf("failed to execute query: %w", queryErr) - } - for _, row := range results { - id, parseErr := parseString(row["checkpoint_id"], "checkpoint_id") - if parseErr != nil { - return parseErr - } - ids = append(ids, id) - } - return nil - }) - if err != nil { - return nil, err - } - return ids, nil -} - -// Delete removes a session's session context. -func (st *store) Delete(ctx context.Context, sessionID string) error { - return withOperationDBClient(st.dbProvider, func(dbClient provider.DBClientInterface) error { - _, err := dbClient.ExecuteContext(ctx, queryDeleteSessionContext, sessionID, st.deploymentID) - if err != nil { - return fmt.Errorf("failed to delete session context: %w", err) - } - return nil - }) -} - -// buildSessionContextFromRow parses a result row into an SessionContext. -func (st *store) buildSessionContextFromRow(row map[string]interface{}) (*SessionContext, error) { - sessionID, err := parseString(row["session_id"], "session_id") - if err != nil { - return nil, err - } - checkpointID, err := parseString(row["checkpoint_id"], "checkpoint_id") - if err != nil { - return nil, err - } - contextVersion, err := parseInt(row["context_version"], "context_version") - if err != nil { - return nil, err - } - payload, err := parseSessionContextPayload(parseNullableString(row["context"])) - if err != nil { - return nil, err - } - - return &SessionContext{ - SessionID: sessionID, - CheckpointID: checkpointID, - RuntimeData: payload.RuntimeData, - AuthUser: payload.AuthUser, - CompletedSteps: payload.CompletedSteps, - ContextVersion: contextVersion, - }, nil -} diff --git a/backend/internal/flow/session/store.go b/backend/internal/flow/session/store.go index 7d1c529882..419b03df07 100644 --- a/backend/internal/flow/session/store.go +++ b/backend/internal/flow/session/store.go @@ -122,6 +122,194 @@ func (st *store) Update(ctx context.Context, s *Session) error { }) } +// CreateContext persists the session context. It rejects payloads exceeding MaxSessionContextBytes. +func (st *store) CreateContext(ctx context.Context, c SessionContext) error { + payload, err := c.serializePayload() + if err != nil { + return err + } + if len(payload) > MaxSessionContextBytes { + return errSessionContextTooLarge + } + + return withOperationDBClient(st.dbProvider, func(dbClient provider.DBClientInterface) error { + _, execErr := dbClient.ExecuteContext(ctx, queryCreateSessionContext, + c.SessionID, st.deploymentID, c.CheckpointID, payload, c.ContextVersion) + if execErr != nil { + return fmt.Errorf("failed to create session context: %w", execErr) + } + return nil + }) +} + +// GetByCheckpoint fetches one checkpoint's session context. +func (st *store) GetByCheckpoint(ctx context.Context, sessionID, + checkpointID string) (*SessionContext, error) { + var result *SessionContext + + err := withOperationDBClient(st.dbProvider, func(dbClient provider.DBClientInterface) error { + results, queryErr := dbClient.QueryContext(ctx, queryGetSessionContextByCheckpoint, + sessionID, st.deploymentID, checkpointID) + if queryErr != nil { + return fmt.Errorf("failed to execute query: %w", queryErr) + } + if len(results) == 0 { + return nil + } + if len(results) != 1 { + return fmt.Errorf("unexpected number of results: %d", len(results)) + } + + c, buildErr := st.buildSessionContextFromRow(results[0]) + if buildErr != nil { + return buildErr + } + result = c + return nil + }) + if err != nil { + return nil, err + } + return result, nil +} + +// ListCheckpointIDs returns the checkpoint ids a session has saved, without decrypting any payload. +func (st *store) ListCheckpointIDs(ctx context.Context, sessionID string) ([]string, error) { + var ids []string + + err := withOperationDBClient(st.dbProvider, func(dbClient provider.DBClientInterface) error { + results, queryErr := dbClient.QueryContext(ctx, queryListCheckpointsBySessionID, sessionID, st.deploymentID) + if queryErr != nil { + return fmt.Errorf("failed to execute query: %w", queryErr) + } + for _, row := range results { + id, parseErr := parseString(row["checkpoint_id"], "checkpoint_id") + if parseErr != nil { + return parseErr + } + ids = append(ids, id) + } + return nil + }) + if err != nil { + return nil, err + } + return ids, nil +} + +// Delete removes a session's session context. +func (st *store) Delete(ctx context.Context, sessionID string) error { + return withOperationDBClient(st.dbProvider, func(dbClient provider.DBClientInterface) error { + _, err := dbClient.ExecuteContext(ctx, queryDeleteSessionContext, sessionID, st.deploymentID) + if err != nil { + return fmt.Errorf("failed to delete session context: %w", err) + } + return nil + }) +} + +// buildSessionContextFromRow parses a result row into an SessionContext. +func (st *store) buildSessionContextFromRow(row map[string]interface{}) (*SessionContext, error) { + sessionID, err := parseString(row["session_id"], "session_id") + if err != nil { + return nil, err + } + checkpointID, err := parseString(row["checkpoint_id"], "checkpoint_id") + if err != nil { + return nil, err + } + contextVersion, err := parseInt(row["context_version"], "context_version") + if err != nil { + return nil, err + } + payload, err := parseSessionContextPayload(parseNullableString(row["context"])) + if err != nil { + return nil, err + } + + return &SessionContext{ + SessionID: sessionID, + CheckpointID: checkpointID, + RuntimeData: payload.RuntimeData, + AuthUser: payload.AuthUser, + CompletedSteps: payload.CompletedSteps, + ContextVersion: contextVersion, + }, nil +} + +// Record inserts or refreshes a participant under the upsert query. +func (st *store) Record(ctx context.Context, p Participant) error { + return withOperationDBClient(st.dbProvider, func(dbClient provider.DBClientInterface) error { + _, err := dbClient.ExecuteContext(ctx, queryUpsertParticipant, + p.SessionID, st.deploymentID, p.AppID, p.FirstJoinedAt, p.LastActiveAt) + if err != nil { + return fmt.Errorf("failed to record session participant: %w", err) + } + return nil + }) +} + +// ListBySessionID returns the participants of a session, oldest first. +func (st *store) ListBySessionID(ctx context.Context, sessionID string) ([]Participant, error) { + var result []Participant + + err := withOperationDBClient(st.dbProvider, func(dbClient provider.DBClientInterface) error { + results, queryErr := dbClient.QueryContext(ctx, queryListParticipantsBySessionID, sessionID, st.deploymentID) + if queryErr != nil { + return fmt.Errorf("failed to execute query: %w", queryErr) + } + for _, row := range results { + p, buildErr := buildParticipantFromRow(row) + if buildErr != nil { + return buildErr + } + result = append(result, p) + } + return nil + }) + if err != nil { + return nil, err + } + return result, nil +} + +// DeleteBySessionID removes all participants of a session. +func (st *store) DeleteBySessionID(ctx context.Context, sessionID string) error { + return withOperationDBClient(st.dbProvider, func(dbClient provider.DBClientInterface) error { + _, err := dbClient.ExecuteContext(ctx, queryDeleteParticipantsBySessionID, sessionID, st.deploymentID) + if err != nil { + return fmt.Errorf("failed to delete session participants: %w", err) + } + return nil + }) +} + +// buildParticipantFromRow maps a database result row into a Participant. +func buildParticipantFromRow(row map[string]interface{}) (Participant, error) { + sessionID, err := parseString(row["session_id"], "session_id") + if err != nil { + return Participant{}, err + } + appID, err := parseString(row["app_id"], "app_id") + if err != nil { + return Participant{}, err + } + firstJoinedAt, err := sysutils.ParseDBTimeField(row["first_joined_at"], "first_joined_at") + if err != nil { + return Participant{}, err + } + lastActiveAt, err := sysutils.ParseDBTimeField(row["last_active_at"], "last_active_at") + if err != nil { + return Participant{}, err + } + return Participant{ + SessionID: sessionID, + AppID: appID, + FirstJoinedAt: firstJoinedAt, + LastActiveAt: lastActiveAt, + }, nil +} + // withOperationDBClient runs fn with an operation database client. SSO sessions are persistent // state that must survive a runtime database flush, so they live in the operation datasource. func withOperationDBClient(dbProvider provider.DBProviderInterface, diff --git a/backend/internal/flow/session/store_constants.go b/backend/internal/flow/session/store_constants.go index 3290d51f1a..15af1b0ad1 100644 --- a/backend/internal/flow/session/store_constants.go +++ b/backend/internal/flow/session/store_constants.go @@ -49,7 +49,7 @@ var ( // rows when that execution has not established one. Used on the fresh join path to attach later // checkpoints to the session an earlier join in the same execution already created. queryGetSessionByExecutionID = model.DBQuery{ - ID: "SSO-SESS-04", + ID: "SSO-SESS-03", Query: `SELECT SESSION_ID, SUBJECT_ID, FLOW_ID, FLOW_VERSION, FLOW_EXECUTION_ID, HANDLE_ID, ` + `AUTHENTICATED_AT, CREATED_AT, LAST_ACTIVE_AT, IDLE_EXPIRES_AT, ABSOLUTE_EXPIRES_AT, STATE, VERSION ` + `FROM "SSO_SESSION" WHERE FLOW_EXECUTION_ID = $1 AND DEPLOYMENT_ID = $2`, @@ -59,7 +59,7 @@ var ( // guard: it only matches when the stored VERSION equals the expected version, and it // bumps VERSION on success. It never touches the session context. queryUpdateSession = model.DBQuery{ - ID: "SSO-SESS-03", + ID: "SSO-SESS-04", Query: `UPDATE "SSO_SESSION" SET FLOW_VERSION = $1, HANDLE_ID = $2, ` + `LAST_ACTIVE_AT = $3, IDLE_EXPIRES_AT = $4, ABSOLUTE_EXPIRES_AT = $5, STATE = $6, ` + `VERSION = VERSION + 1, UPDATED_AT = CURRENT_TIMESTAMP ` + @@ -70,7 +70,7 @@ var ( // (re-execution or a concurrent request) overwrites it rather than erroring on the primary key. // The ON CONFLICT ... DO UPDATE form is valid in both PostgreSQL and SQLite. queryCreateSessionContext = model.DBQuery{ - ID: "SSO-SESS-AC-01", + ID: "SSO-SESS-05", Query: `INSERT INTO "SSO_SESSION_CONTEXT" (SESSION_ID, DEPLOYMENT_ID, CHECKPOINT_ID, CONTEXT, ` + `CONTEXT_VERSION) VALUES ($1, $2, $3, $4, $5) ` + `ON CONFLICT (SESSION_ID, DEPLOYMENT_ID, CHECKPOINT_ID) DO UPDATE SET ` + @@ -79,21 +79,21 @@ var ( // queryGetSessionContextByCheckpoint fetches one checkpoint's session context for a session. queryGetSessionContextByCheckpoint = model.DBQuery{ - ID: "SSO-SESS-AC-02", + ID: "SSO-SESS-06", Query: `SELECT SESSION_ID, CHECKPOINT_ID, CONTEXT, CONTEXT_VERSION FROM "SSO_SESSION_CONTEXT" ` + `WHERE SESSION_ID = $1 AND DEPLOYMENT_ID = $2 AND CHECKPOINT_ID = $3`, } // queryDeleteSessionContext removes all of a session's checkpoint contexts. queryDeleteSessionContext = model.DBQuery{ - ID: "SSO-SESS-AC-03", + ID: "SSO-SESS-07", Query: `DELETE FROM "SSO_SESSION_CONTEXT" WHERE SESSION_ID = $1 AND DEPLOYMENT_ID = $2`, } // queryListCheckpointsBySessionID returns the checkpoint ids a session has saved. It is the // existence check the SSO-Check node uses to decide availability without decrypting any context. queryListCheckpointsBySessionID = model.DBQuery{ - ID: "SSO-SESS-AC-04", + ID: "SSO-SESS-08", Query: `SELECT CHECKPOINT_ID FROM "SSO_SESSION_CONTEXT" ` + `WHERE SESSION_ID = $1 AND DEPLOYMENT_ID = $2`, } @@ -102,7 +102,7 @@ var ( // LAST_ACTIVE_AT (but preserving FIRST_JOINED_AT) when the application has already joined. The // ON CONFLICT ... DO UPDATE form is valid in both PostgreSQL and SQLite. queryUpsertParticipant = model.DBQuery{ - ID: "SSO-SESS-PART-01", + ID: "SSO-SESS-09", Query: `INSERT INTO "SSO_SESSION_PARTICIPANT" ` + `(SESSION_ID, DEPLOYMENT_ID, APP_ID, FIRST_JOINED_AT, LAST_ACTIVE_AT) ` + `VALUES ($1, $2, $3, $4, $5) ` + @@ -112,14 +112,14 @@ var ( // queryListParticipantsBySessionID returns the applications that have joined a session, oldest // first. queryListParticipantsBySessionID = model.DBQuery{ - ID: "SSO-SESS-PART-02", + ID: "SSO-SESS-10", Query: `SELECT SESSION_ID, APP_ID, FIRST_JOINED_AT, LAST_ACTIVE_AT FROM "SSO_SESSION_PARTICIPANT" ` + `WHERE SESSION_ID = $1 AND DEPLOYMENT_ID = $2 ORDER BY FIRST_JOINED_AT`, } // queryDeleteParticipantsBySessionID removes all participants of a session. queryDeleteParticipantsBySessionID = model.DBQuery{ - ID: "SSO-SESS-PART-03", + ID: "SSO-SESS-11", Query: `DELETE FROM "SSO_SESSION_PARTICIPANT" WHERE SESSION_ID = $1 AND DEPLOYMENT_ID = $2`, } )