package postgres import ( "context" "encoding/json" "errors" "fmt" "github.com/jackc/pgx/v5" "knowledge-governance/internal/governance" ) var ErrScopeTargetNotFound = errors.New("scope target not found") func (s *Store) ListScopeRules(ctx context.Context, connectionID string) ([]governance.ScopeRule, error) { if connectionID == "" { return nil, ErrScopeTargetNotFound } var exists bool if err := s.db.QueryRow(ctx, `SELECT EXISTS(SELECT 1 FROM source_connections WHERE id=$1)`, connectionID).Scan(&exists); err != nil { return nil, fmt.Errorf("inspect scope connection: %w", err) } if !exists { return nil, ErrScopeTargetNotFound } rows, err := s.db.Query(ctx, ` SELECT 'directory', source_connection_id, directory_path, '', visibility, updated_by, reason, updated_at FROM directory_scope_rules WHERE source_connection_id=$1 UNION ALL SELECT 'item', item.source_connection_id, '', override.source_item_id, override.visibility, override.updated_by, override.reason, override.updated_at FROM source_item_scope_overrides override JOIN source_items item ON item.id=override.source_item_id WHERE item.source_connection_id=$1 ORDER BY 1, 3, 4`, connectionID) if err != nil { return nil, fmt.Errorf("list scope rules: %w", err) } defer rows.Close() rules := make([]governance.ScopeRule, 0) for rows.Next() { var rule governance.ScopeRule if err := rows.Scan(&rule.TargetType, &rule.ConnectionID, &rule.DirectoryPath, &rule.SourceItemID, &rule.Visibility, &rule.UpdatedBy, &rule.Reason, &rule.UpdatedAt); err != nil { return nil, fmt.Errorf("scan scope rule: %w", err) } rules = append(rules, rule) } if err := rows.Err(); err != nil { return nil, fmt.Errorf("iterate scope rules: %w", err) } return rules, nil } func (s *Store) MutateScopeRule(ctx context.Context, request governance.ScopeRuleMutation) (governance.ScopeRule, error) { if err := request.Validate(); err != nil { return governance.ScopeRule{}, err } beginner, ok := s.db.(interface { Begin(context.Context) (pgx.Tx, error) }) if !ok { return governance.ScopeRule{}, errors.New("scope rule transactions are unavailable") } tx, err := beginner.Begin(ctx) if err != nil { return governance.ScopeRule{}, fmt.Errorf("begin scope rule mutation: %w", err) } defer func() { _ = tx.Rollback(ctx) }() rule, err := mutateScopeRuleTx(ctx, tx, request) if err != nil { return governance.ScopeRule{}, err } if err := tx.Commit(ctx); err != nil { return governance.ScopeRule{}, fmt.Errorf("commit scope rule: %w", err) } return rule, nil } func (s *Store) MutateScopeRuleWithOperation(ctx context.Context, request governance.ScopeRuleMutation, operationID string) (governance.ScopeRule, bool, error) { if err := request.Validate(); err != nil { return governance.ScopeRule{}, false, err } fingerprint, err := operationFingerprint(struct { TargetType governance.ScopeTargetType `json:"target_type"` ConnectionID string `json:"connection_id"` DirectoryPath string `json:"directory_path"` SourceItemID string `json:"source_item_id"` Visibility governance.Visibility `json:"visibility"` OperatorRef string `json:"operator_ref"` Reason string `json:"reason"` Clear bool `json:"clear"` }{request.TargetType, request.ConnectionID, request.DirectoryPath, request.SourceItemID, request.Visibility, request.OperatorRef, request.Reason, request.Clear}) if err != nil { return governance.ScopeRule{}, false, err } resultJSON, replayed, err := s.executeOperation(ctx, "scope.rule.mutate", operationID, fingerprint, request.Now, func(tx pgx.Tx) (json.RawMessage, error) { rule, err := mutateScopeRuleTx(ctx, tx, request) if err != nil { return nil, err } return json.Marshal(rule) }) if err != nil { return governance.ScopeRule{}, false, err } var rule governance.ScopeRule if err := json.Unmarshal(resultJSON, &rule); err != nil { return governance.ScopeRule{}, false, fmt.Errorf("decode stored scope rule result: %w", err) } return rule, replayed, nil } func mutateScopeRuleTx(ctx context.Context, tx pgx.Tx, request governance.ScopeRuleMutation) (governance.ScopeRule, error) { var targetExists bool var err error if request.TargetType == governance.ScopeTargetDirectory { err = tx.QueryRow(ctx, `SELECT EXISTS(SELECT 1 FROM source_connections WHERE id=$1)`, request.ConnectionID).Scan(&targetExists) } else { err = tx.QueryRow(ctx, `SELECT EXISTS(SELECT 1 FROM source_items WHERE id=$1)`, request.SourceItemID).Scan(&targetExists) } if err != nil { return governance.ScopeRule{}, fmt.Errorf("inspect scope target: %w", err) } if !targetExists { return governance.ScopeRule{}, ErrScopeTargetNotFound } var previous *string var previousValue string if request.TargetType == governance.ScopeTargetDirectory { err = tx.QueryRow(ctx, `SELECT visibility FROM directory_scope_rules WHERE source_connection_id=$1 AND directory_path=$2 FOR UPDATE`, request.ConnectionID, request.DirectoryPath).Scan(&previousValue) } else { err = tx.QueryRow(ctx, `SELECT visibility FROM source_item_scope_overrides WHERE source_item_id=$1 FOR UPDATE`, request.SourceItemID).Scan(&previousValue) } if err == nil { previous = &previousValue } else if !errors.Is(err, pgx.ErrNoRows) { return governance.ScopeRule{}, fmt.Errorf("lock scope rule: %w", err) } if request.TargetType == governance.ScopeTargetDirectory { if request.Clear { _, err = tx.Exec(ctx, `DELETE FROM directory_scope_rules WHERE source_connection_id=$1 AND directory_path=$2`, request.ConnectionID, request.DirectoryPath) } else { _, err = tx.Exec(ctx, `INSERT INTO directory_scope_rules (source_connection_id,directory_path,visibility,updated_by,reason,created_at,updated_at) VALUES ($1,$2,$3,$4,$5,$6,$6) ON CONFLICT (source_connection_id,directory_path) DO UPDATE SET visibility=EXCLUDED.visibility,updated_by=EXCLUDED.updated_by,reason=EXCLUDED.reason,updated_at=EXCLUDED.updated_at`, request.ConnectionID, request.DirectoryPath, request.Visibility, request.OperatorRef, request.Reason, request.Now) } } else if request.Clear { _, err = tx.Exec(ctx, `DELETE FROM source_item_scope_overrides WHERE source_item_id=$1`, request.SourceItemID) } else { _, err = tx.Exec(ctx, `INSERT INTO source_item_scope_overrides (source_item_id,visibility,updated_by,reason,created_at,updated_at) VALUES ($1,$2,$3,$4,$5,$5) ON CONFLICT (source_item_id) DO UPDATE SET visibility=EXCLUDED.visibility,updated_by=EXCLUDED.updated_by,reason=EXCLUDED.reason,updated_at=EXCLUDED.updated_at`, request.SourceItemID, request.Visibility, request.OperatorRef, request.Reason, request.Now) } if err != nil { return governance.ScopeRule{}, fmt.Errorf("write scope rule: %w", err) } eventID, err := newOpaqueID("se-") if err != nil { return governance.ScopeRule{}, err } var decided any if !request.Clear { decided = string(request.Visibility) } _, err = tx.Exec(ctx, `INSERT INTO scope_change_events (id,target_type,source_connection_id,directory_path,source_item_id,action,previous_visibility,decided_visibility,operator_ref,reason,created_at) VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11)`, eventID, request.TargetType, nullableString(request.ConnectionID), request.DirectoryPath, nullableString(request.SourceItemID), map[bool]string{true: "clear", false: "set"}[request.Clear], previous, decided, request.OperatorRef, request.Reason, request.Now) if err != nil { return governance.ScopeRule{}, fmt.Errorf("audit scope rule: %w", err) } return governance.ScopeRule{TargetType: request.TargetType, ConnectionID: request.ConnectionID, DirectoryPath: request.DirectoryPath, SourceItemID: request.SourceItemID, Visibility: request.Visibility, UpdatedBy: request.OperatorRef, Reason: request.Reason, UpdatedAt: request.Now}, nil } func nullableString(value string) any { if value == "" { return nil } return value }