package storage import ( "context" "encoding/json" "errors" "fmt" "strconv" "strings" "time" "github.com/jackc/pgx/v5" "github.com/jackc/pgx/v5/pgconn" "archivdms/internal/matching" ) // ErrWorkflowNotFound is returned when a workflow lookup, update or delete does // not match any row owned by the caller's tenant. var ErrWorkflowNotFound = errors.New("storage: workflow not found or not owned by tenant") // ErrDuplicateWorkflowName is returned when a tenant already has a workflow // with the same name (UNIQUE(tenant_id, name)). var ErrDuplicateWorkflowName = errors.New("storage: workflow with this name already exists for tenant") // ErrInvalidConditionTree is returned when a supplied condition_tree is // malformed or exceeds the MVP nesting-depth guardrail. var ErrInvalidConditionTree = errors.New("storage: invalid workflow condition tree") // ErrInvalidWorkflowAction is returned when a supplied workflow action has an // unknown action_type or a malformed action_config for its type. var ErrInvalidWorkflowAction = errors.New("storage: invalid workflow action") // Supported workflow trigger types (mirrors the trigger_type CHECK). const WorkflowTriggerOnUpload = "on_upload" // Supported workflow action types (mirrors the action_type CHECK). const ( ActionAddTag = "add_tag" ActionSetDocType = "set_doc_type" ActionSetCorrespondent = "set_correspondent" ActionApplyClassificationTmpl = "apply_classification_template" ActionSetCustomField = "set_custom_field" ) // Leaf fields supported in a condition tree (MVP). const ( FieldOCRText = "ocr_text" FieldTitle = "title" FieldDocType = "doc_type" FieldCorrespondent = "correspondent" FieldSource = "source" ) // Boolean operators in a condition tree. const ( OpAnd = "and" OpOr = "or" ) // ConditionNode is one node of a workflow condition tree. It is either a // boolean group (Op + Children set) or a leaf (Field + Algorithm + Pattern // set). The two forms are mutually exclusive. Parsed/validated on write and // stored both as the typed struct (validation) and raw JSONB (DB column). type ConditionNode struct { // Group form. Op string `json:"op,omitempty"` Children []ConditionNode `json:"children,omitempty"` // Leaf form. Field string `json:"field,omitempty"` Algorithm string `json:"algorithm,omitempty"` Pattern string `json:"pattern,omitempty"` CaseSensitive bool `json:"case_sensitive,omitempty"` } // isGroup reports whether the node is a boolean group (has an op). func (n ConditionNode) isGroup() bool { return n.Op != "" } // Workflow is a tenant-scoped automation rule evaluated at a trigger point // (MVP: on_upload). A matched workflow runs its ordered actions against the // document. Every evaluation is recorded in workflow_runs for GoBD // traceability (document-scoped audit trail parallel to internal/audit). type Workflow struct { ID int64 `json:"id"` TenantID int64 `json:"tenant_id"` Name string `json:"name"` Enabled bool `json:"enabled"` TriggerType string `json:"trigger_type"` ConditionTree ConditionNode `json:"condition_tree"` Priority int `json:"priority"` CreatedBy *int64 `json:"created_by,omitempty"` CreatedAt time.Time `json:"created_at"` UpdatedAt time.Time `json:"updated_at"` Actions []WorkflowAction `json:"actions"` } // WorkflowAction is one ordered action of a workflow, with its type-specific // configuration carried as raw JSON (validated on write). type WorkflowAction struct { ID int64 `json:"id"` WorkflowID int64 `json:"workflow_id"` StepOrder int `json:"step_order"` ActionType string `json:"action_type"` ActionConfig json.RawMessage `json:"action_config"` } // WorkflowActionInput is one supplied action in a bulk PUT (step_order is the // slice index, assigned by SetWorkflowActions). type WorkflowActionInput struct { ActionType string `json:"action_type"` ActionConfig json.RawMessage `json:"action_config"` } // CreateWorkflowRequest holds create parameters for a workflow. type CreateWorkflowRequest struct { Name string Enabled bool TriggerType string ConditionTree json.RawMessage Priority int CreatedBy *int64 } // UpdateWorkflowRequest holds update parameters for a workflow. type UpdateWorkflowRequest struct { Name string Enabled bool TriggerType string ConditionTree json.RawMessage Priority int } // WorkflowRun is one recorded evaluation of a workflow against a document. type WorkflowRun struct { ID int64 `json:"id"` WorkflowID int64 `json:"workflow_id"` TenantID int64 `json:"tenant_id"` DocumentID *int64 `json:"document_id,omitempty"` TriggeredAt time.Time `json:"triggered_at"` Matched bool `json:"matched"` ActionsApplied json.RawMessage `json:"actions_applied,omitempty"` Error string `json:"error,omitempty"` } // WorkflowTestResult is the dry-run outcome of TestWorkflow: whether the tree // matched, the per-leaf evaluation, and which actions WOULD run (never // executed). type WorkflowTestResult struct { WorkflowID int64 `json:"workflow_id"` Matched bool `json:"matched"` Leaves []LeafEvalResult `json:"leaves"` ActionsToRun []WorkflowAction `json:"actions_to_run"` } // LeafEvalResult records the evaluation of a single leaf condition. type LeafEvalResult struct { Field string `json:"field"` Algorithm string `json:"algorithm"` Pattern string `json:"pattern"` Value string `json:"value"` Matched bool `json:"matched"` } // initWorkflowsSchema creates the workflows / workflow_actions / workflow_runs // tables. Idempotent, called from (*Store).initSchema AFTER // initClassificationTemplatesSchema (a workflow action can reference a // classification template). Documented (not executed) in // migrations/012_workflows.sql. func (s *Store) initWorkflowsSchema(ctx context.Context) error { _, err := s.db.Exec(ctx, ` CREATE TABLE IF NOT EXISTS workflows ( id BIGSERIAL PRIMARY KEY, tenant_id BIGINT NOT NULL, name TEXT NOT NULL, enabled BOOLEAN NOT NULL DEFAULT true, trigger_type TEXT NOT NULL CHECK (trigger_type IN ('on_upload')), condition_tree JSONB NOT NULL, priority INT NOT NULL DEFAULT 100, created_by BIGINT, created_at TIMESTAMPTZ NOT NULL DEFAULT now(), updated_at TIMESTAMPTZ NOT NULL DEFAULT now(), UNIQUE(tenant_id, name) ); CREATE INDEX IF NOT EXISTS idx_workflows_tenant ON workflows(tenant_id); CREATE TABLE IF NOT EXISTS workflow_actions ( id BIGSERIAL PRIMARY KEY, workflow_id BIGINT NOT NULL REFERENCES workflows(id) ON DELETE CASCADE, step_order INT NOT NULL, action_type TEXT NOT NULL CHECK (action_type IN ('add_tag','set_doc_type','set_correspondent','apply_classification_template','set_custom_field')), action_config JSONB NOT NULL DEFAULT '{}' ); CREATE INDEX IF NOT EXISTS idx_workflow_actions_workflow ON workflow_actions(workflow_id, step_order); CREATE TABLE IF NOT EXISTS workflow_runs ( id BIGSERIAL PRIMARY KEY, workflow_id BIGINT NOT NULL REFERENCES workflows(id) ON DELETE CASCADE, tenant_id BIGINT NOT NULL, document_id BIGINT, triggered_at TIMESTAMPTZ NOT NULL DEFAULT now(), matched BOOLEAN NOT NULL, actions_applied JSONB, error TEXT ); CREATE INDEX IF NOT EXISTS idx_workflow_runs_workflow ON workflow_runs(workflow_id); CREATE INDEX IF NOT EXISTS idx_workflow_runs_document ON workflow_runs(document_id); `) if err != nil { return fmt.Errorf("storage: create workflows tables: %w", err) } return nil } // --- condition tree validation -------------------------------------------- // maxConditionDepth is the MVP nesting guardrail: a top-level op plus one level // of nested op groups (depth 2). Deeper trees are rejected — an MVP guardrail, // not a hard architectural limit. const maxConditionDepth = 2 var validConditionFields = map[string]bool{ FieldOCRText: true, FieldTitle: true, FieldDocType: true, FieldCorrespondent: true, FieldSource: true, } var validConditionAlgorithms = map[string]bool{ matching.AlgorithmAny: true, matching.AlgorithmAll: true, matching.AlgorithmExact: true, matching.AlgorithmRegex: true, matching.AlgorithmFuzzy: true, } // parseConditionTree unmarshals and validates a raw condition tree into the // typed ConditionNode, enforcing the depth guardrail and leaf/group shape. func parseConditionTree(raw json.RawMessage) (ConditionNode, error) { if len(raw) == 0 { return ConditionNode{}, fmt.Errorf("%w: empty", ErrInvalidConditionTree) } var root ConditionNode if err := json.Unmarshal(raw, &root); err != nil { return ConditionNode{}, fmt.Errorf("%w: %v", ErrInvalidConditionTree, err) } if err := validateConditionNode(root, 1); err != nil { return ConditionNode{}, err } return root, nil } // validateConditionNode recursively validates a node. depth starts at 1 for // the root; a group's children are at depth+1 and may not exceed // maxConditionDepth. func validateConditionNode(n ConditionNode, depth int) error { if n.isGroup() { if n.Op != OpAnd && n.Op != OpOr { return fmt.Errorf("%w: unknown op %q", ErrInvalidConditionTree, n.Op) } if n.Field != "" || n.Algorithm != "" || n.Pattern != "" { return fmt.Errorf("%w: node has both group op and leaf fields", ErrInvalidConditionTree) } if len(n.Children) == 0 { return fmt.Errorf("%w: group %q has no children", ErrInvalidConditionTree, n.Op) } if depth >= maxConditionDepth { // Children of this group would exceed the guardrail unless they are // all leaves. Enforce: at depth == maxConditionDepth no child may be // a group. for _, c := range n.Children { if c.isGroup() { return fmt.Errorf("%w: nesting deeper than %d levels", ErrInvalidConditionTree, maxConditionDepth) } } } for _, c := range n.Children { if err := validateConditionNode(c, depth+1); err != nil { return err } } return nil } // Leaf. if !validConditionFields[n.Field] { return fmt.Errorf("%w: unknown field %q", ErrInvalidConditionTree, n.Field) } if !validConditionAlgorithms[n.Algorithm] { return fmt.Errorf("%w: unknown algorithm %q", ErrInvalidConditionTree, n.Algorithm) } if strings.TrimSpace(n.Pattern) == "" { return fmt.Errorf("%w: leaf field %q has empty pattern", ErrInvalidConditionTree, n.Field) } return nil } // validateActionConfig validates one action's type and its JSON config shape. func validateActionConfig(actionType string, cfg json.RawMessage) error { switch actionType { case ActionAddTag: var c struct { TagID int64 `json:"tag_id"` } if err := json.Unmarshal(nonNilJSON(cfg), &c); err != nil || c.TagID == 0 { return fmt.Errorf("%w: %s requires tag_id", ErrInvalidWorkflowAction, actionType) } case ActionSetDocType: var c struct { DocTypeID int64 `json:"doc_type_id"` } if err := json.Unmarshal(nonNilJSON(cfg), &c); err != nil || c.DocTypeID == 0 { return fmt.Errorf("%w: %s requires doc_type_id", ErrInvalidWorkflowAction, actionType) } case ActionSetCorrespondent: var c struct { CorrespondentID int64 `json:"correspondent_id"` } if err := json.Unmarshal(nonNilJSON(cfg), &c); err != nil || c.CorrespondentID == 0 { return fmt.Errorf("%w: %s requires correspondent_id", ErrInvalidWorkflowAction, actionType) } case ActionApplyClassificationTmpl: var c struct { TemplateID int64 `json:"template_id"` } if err := json.Unmarshal(nonNilJSON(cfg), &c); err != nil || c.TemplateID == 0 { return fmt.Errorf("%w: %s requires template_id", ErrInvalidWorkflowAction, actionType) } case ActionSetCustomField: var c struct { FieldID int64 `json:"field_id"` } if err := json.Unmarshal(nonNilJSON(cfg), &c); err != nil || c.FieldID == 0 { return fmt.Errorf("%w: %s requires field_id", ErrInvalidWorkflowAction, actionType) } default: return fmt.Errorf("%w: unknown action_type %q", ErrInvalidWorkflowAction, actionType) } return nil } // nonNilJSON returns raw, or an empty JSON object when raw is nil/empty, so // json.Unmarshal does not fail on a missing action_config. func nonNilJSON(raw json.RawMessage) json.RawMessage { if len(raw) == 0 { return json.RawMessage(`{}`) } return raw } // --- CRUD ------------------------------------------------------------------ const workflowCols = `id, tenant_id, name, enabled, trigger_type, condition_tree, priority, created_by, created_at, updated_at` // scanWorkflow scans a workflow row (without actions) and parses its raw // condition_tree JSONB into the typed struct. func scanWorkflow(row interface { Scan(dest ...any) error }) (*Workflow, error) { var w Workflow var rawTree []byte if err := row.Scan(&w.ID, &w.TenantID, &w.Name, &w.Enabled, &w.TriggerType, &rawTree, &w.Priority, &w.CreatedBy, &w.CreatedAt, &w.UpdatedAt); err != nil { return nil, err } // Best-effort parse: a stored tree was validated on write, so a parse error // here is a data-integrity anomaly — surface it rather than silently drop. if len(rawTree) > 0 { if err := json.Unmarshal(rawTree, &w.ConditionTree); err != nil { return nil, fmt.Errorf("storage: unmarshal stored condition_tree: %w", err) } } w.Actions = make([]WorkflowAction, 0) return &w, nil } // CreateWorkflow inserts a new workflow (without actions — those are set via // SetWorkflowActions). It validates the condition tree depth/shape and stores // both the validated tree (as raw JSONB) and the typed struct on the return. func (s *Store) CreateWorkflow(ctx context.Context, tenantID int64, req CreateWorkflowRequest) (*Workflow, error) { if req.TriggerType == "" { req.TriggerType = WorkflowTriggerOnUpload } if req.TriggerType != WorkflowTriggerOnUpload { return nil, fmt.Errorf("%w: unsupported trigger_type %q", ErrInvalidConditionTree, req.TriggerType) } tree, err := parseConditionTree(req.ConditionTree) if err != nil { return nil, err } row := s.db.QueryRow(ctx, ` INSERT INTO workflows (tenant_id, name, enabled, trigger_type, condition_tree, priority, created_by) VALUES ($1, $2, $3, $4, $5, $6, $7) RETURNING `+workflowCols, tenantID, req.Name, req.Enabled, req.TriggerType, []byte(req.ConditionTree), req.Priority, req.CreatedBy) w, err := scanWorkflow(row) if err != nil { var pgErr *pgconn.PgError if errors.As(err, &pgErr) && pgErr.Code == "23505" { return nil, ErrDuplicateWorkflowName } return nil, fmt.Errorf("storage: create workflow: %w", err) } w.ConditionTree = tree return w, nil } // ListWorkflows returns all workflows for a tenant (without actions), ordered // by priority then name. Returns a non-nil (possibly empty) slice. func (s *Store) ListWorkflows(ctx context.Context, tenantID int64) ([]Workflow, error) { rows, err := s.db.Query(ctx, `SELECT `+workflowCols+` FROM workflows WHERE tenant_id = $1 ORDER BY priority ASC, name ASC`, tenantID) if err != nil { return nil, fmt.Errorf("storage: list workflows: %w", err) } defer rows.Close() out := make([]Workflow, 0) for rows.Next() { w, err := scanWorkflow(rows) if err != nil { return nil, fmt.Errorf("storage: scan workflow: %w", err) } out = append(out, *w) } return out, rows.Err() } // GetWorkflow returns one workflow resolved with its ordered actions, scoped to // tenant ownership. func (s *Store) GetWorkflow(ctx context.Context, id, tenantID int64) (*Workflow, error) { row := s.db.QueryRow(ctx, `SELECT `+workflowCols+` FROM workflows WHERE id = $1 AND tenant_id = $2`, id, tenantID) w, err := scanWorkflow(row) if err != nil { if errors.Is(err, pgx.ErrNoRows) { return nil, ErrWorkflowNotFound } return nil, fmt.Errorf("storage: get workflow: %w", err) } actions, err := s.listWorkflowActions(ctx, id) if err != nil { return nil, err } w.Actions = actions return w, nil } // listWorkflowActions returns the actions of a workflow ordered by step_order. func (s *Store) listWorkflowActions(ctx context.Context, workflowID int64) ([]WorkflowAction, error) { rows, err := s.db.Query(ctx, ` SELECT id, workflow_id, step_order, action_type, action_config FROM workflow_actions WHERE workflow_id = $1 ORDER BY step_order ASC`, workflowID) if err != nil { return nil, fmt.Errorf("storage: list workflow actions: %w", err) } defer rows.Close() out := make([]WorkflowAction, 0) for rows.Next() { var a WorkflowAction var cfg []byte if err := rows.Scan(&a.ID, &a.WorkflowID, &a.StepOrder, &a.ActionType, &cfg); err != nil { return nil, fmt.Errorf("storage: scan workflow action: %w", err) } a.ActionConfig = json.RawMessage(cfg) out = append(out, a) } return out, rows.Err() } // UpdateWorkflow updates a workflow's core attributes (not its actions), scoped // to tenant ownership. Re-validates the condition tree. func (s *Store) UpdateWorkflow(ctx context.Context, id, tenantID int64, req UpdateWorkflowRequest) error { if req.TriggerType == "" { req.TriggerType = WorkflowTriggerOnUpload } if req.TriggerType != WorkflowTriggerOnUpload { return fmt.Errorf("%w: unsupported trigger_type %q", ErrInvalidConditionTree, req.TriggerType) } if _, err := parseConditionTree(req.ConditionTree); err != nil { return err } tag, err := s.db.Exec(ctx, ` UPDATE workflows SET name = $1, enabled = $2, trigger_type = $3, condition_tree = $4, priority = $5, updated_at = now() WHERE id = $6 AND tenant_id = $7 `, req.Name, req.Enabled, req.TriggerType, []byte(req.ConditionTree), req.Priority, id, tenantID) if err != nil { var pgErr *pgconn.PgError if errors.As(err, &pgErr) && pgErr.Code == "23505" { return ErrDuplicateWorkflowName } return fmt.Errorf("storage: update workflow: %w", err) } if tag.RowsAffected() == 0 { return ErrWorkflowNotFound } return nil } // DeleteWorkflow deletes a workflow (cascades to its actions / runs), scoped to // tenant ownership. func (s *Store) DeleteWorkflow(ctx context.Context, id, tenantID int64) error { tag, err := s.db.Exec(ctx, `DELETE FROM workflows WHERE id = $1 AND tenant_id = $2`, id, tenantID) if err != nil { return fmt.Errorf("storage: delete workflow: %w", err) } if tag.RowsAffected() == 0 { return ErrWorkflowNotFound } return nil } // workflowOwned returns true if the workflow belongs to the tenant. func (s *Store) workflowOwned(ctx context.Context, workflowID, tenantID int64) (bool, error) { var ok bool err := s.db.QueryRow(ctx, `SELECT EXISTS(SELECT 1 FROM workflows WHERE id = $1 AND tenant_id = $2)`, workflowID, tenantID).Scan(&ok) if err != nil { return false, fmt.Errorf("storage: check workflow ownership: %w", err) } return ok, nil } // SetWorkflowActions replaces the complete ordered set of actions on a workflow // (bulk PUT). step_order is the slice index. Each action's config is validated // against its type. Delete-all + insert in one transaction (mirrors // SetTemplateFieldDefaults). Scoped to tenant ownership of the workflow. func (s *Store) SetWorkflowActions(ctx context.Context, workflowID, tenantID int64, actions []WorkflowActionInput) error { owns, err := s.workflowOwned(ctx, workflowID, tenantID) if err != nil { return err } if !owns { return ErrWorkflowNotFound } for _, a := range actions { if err := validateActionConfig(a.ActionType, a.ActionConfig); err != nil { return err } } tx, err := s.db.Begin(ctx) if err != nil { return fmt.Errorf("storage: begin set workflow actions: %w", err) } defer tx.Rollback(ctx) if _, err := tx.Exec(ctx, `DELETE FROM workflow_actions WHERE workflow_id = $1`, workflowID); err != nil { return fmt.Errorf("storage: clear workflow actions: %w", err) } for i, a := range actions { if _, err := tx.Exec(ctx, ` INSERT INTO workflow_actions (workflow_id, step_order, action_type, action_config) VALUES ($1, $2, $3, $4) `, workflowID, i, a.ActionType, []byte(nonNilJSON(a.ActionConfig))); err != nil { return fmt.Errorf("storage: insert workflow action: %w", err) } } if err := tx.Commit(ctx); err != nil { return fmt.Errorf("storage: commit set workflow actions: %w", err) } return nil } // --- evaluation ------------------------------------------------------------ // resolveLeafValue returns the document's value for a leaf field. For doc_type // and correspondent it resolves the taxonomy entity's name when the *_id is // set, falling back to the deprecated free-text column. func (s *Store) resolveLeafValue(ctx context.Context, field string, doc *Document) (string, error) { switch field { case FieldOCRText: return doc.OCRText, nil case FieldTitle: return doc.Title, nil case FieldSource: return doc.Source, nil case FieldDocType: if doc.DocTypeID != nil { name, err := s.taxonomyName(ctx, "document_types", *doc.DocTypeID, doc.TenantID) if err != nil { return "", err } if name != "" { return name, nil } } return doc.DocType, nil case FieldCorrespondent: if doc.CorrespondentID != nil { name, err := s.taxonomyName(ctx, "correspondents", *doc.CorrespondentID, doc.TenantID) if err != nil { return "", err } if name != "" { return name, nil } } return doc.Correspondent, nil default: return "", nil } } // taxonomyName resolves a taxonomy entity's name by id, scoped to tenant. // Returns "" (no error) when no row matches. func (s *Store) taxonomyName(ctx context.Context, kind string, id, tenantID int64) (string, error) { table, err := taxonomyTable(kind) if err != nil { return "", err } var name string err = s.db.QueryRow(ctx, fmt.Sprintf(`SELECT name FROM %s WHERE id = $1 AND tenant_id = $2`, table), id, tenantID).Scan(&name) if err != nil { if errors.Is(err, pgx.ErrNoRows) { return "", nil } return "", fmt.Errorf("storage: resolve %s name: %w", kind, err) } return name, nil } // evalNode recursively evaluates a condition node against a document, appending // per-leaf results to *leaves (nil to skip collection). When leaves is non-nil // (test endpoint) all children are walked so every leaf is reported; the group // result itself is still computed honestly. func (s *Store) evalNode(ctx context.Context, n ConditionNode, doc *Document, leaves *[]LeafEvalResult) (bool, error) { if n.isGroup() { if n.Op == OpAnd { result := true for _, c := range n.Children { ok, err := s.evalNode(ctx, c, doc, leaves) if err != nil { return false, err } if !ok { result = false if leaves == nil { return false, nil // short-circuit when not collecting } } } return result, nil } // OR. result := false for _, c := range n.Children { ok, err := s.evalNode(ctx, c, doc, leaves) if err != nil { return false, err } if ok { result = true if leaves == nil { return true, nil // short-circuit when not collecting } } } return result, nil } // Leaf. value, err := s.resolveLeafValue(ctx, n.Field, doc) if err != nil { return false, err } matched := matching.Match(n.Algorithm, n.Pattern, n.CaseSensitive, value) if leaves != nil { *leaves = append(*leaves, LeafEvalResult{ Field: n.Field, Algorithm: n.Algorithm, Pattern: n.Pattern, Value: value, Matched: matched, }) } return matched, nil } // EvaluateWorkflow walks the workflow's condition tree against a document and // reports whether it matched. Leaf conditions reuse the matching-engine // algorithms (internal/matching) against the resolved field value. func (s *Store) EvaluateWorkflow(ctx context.Context, workflow *Workflow, doc *Document) (bool, error) { return s.evalNode(ctx, workflow.ConditionTree, doc, nil) } // --- execution ------------------------------------------------------------- // RunWorkflowsForDocument loads all enabled workflows of the tenant with the // given trigger type (ordered by priority ASC), evaluates each and runs the // ordered actions of matched ones. Every workflow evaluated produces a // workflow_runs row (matched true/false, actions_applied summary, error). A // failed action never aborts the document upload — it is recorded and the loop // continues (mirrors the "never fail the upload" philosophy of the OCR / // auto-assign steps in storeUploadedFile). func (s *Store) RunWorkflowsForDocument(ctx context.Context, tenantID int64, doc *Document, triggerType string) error { rows, err := s.db.Query(ctx, `SELECT `+workflowCols+` FROM workflows WHERE tenant_id = $1 AND enabled = true AND trigger_type = $2 ORDER BY priority ASC, id ASC`, tenantID, triggerType) if err != nil { return fmt.Errorf("storage: load workflows for document: %w", err) } workflows := make([]Workflow, 0) for rows.Next() { w, err := scanWorkflow(rows) if err != nil { rows.Close() return fmt.Errorf("storage: scan workflow for document: %w", err) } workflows = append(workflows, *w) } rows.Close() if err := rows.Err(); err != nil { return fmt.Errorf("storage: iterate workflows for document: %w", err) } for i := range workflows { w := &workflows[i] matched, err := s.EvaluateWorkflow(ctx, w, doc) if err != nil { s.recordWorkflowRun(ctx, w.ID, tenantID, &doc.ID, false, nil, err.Error()) continue } if !matched { s.recordWorkflowRun(ctx, w.ID, tenantID, &doc.ID, false, nil, "") continue } actions, err := s.listWorkflowActions(ctx, w.ID) if err != nil { s.recordWorkflowRun(ctx, w.ID, tenantID, &doc.ID, true, nil, err.Error()) continue } applied, runErr := s.executeWorkflowActions(ctx, tenantID, doc, actions) var errStr string if runErr != nil { errStr = runErr.Error() } s.recordWorkflowRun(ctx, w.ID, tenantID, &doc.ID, true, applied, errStr) } return nil } // executeWorkflowActions runs a workflow's ordered actions against a document. // A failed action is recorded in the returned summary and as the returned // error (first failure) but does not stop subsequent actions — best-effort, // each action's outcome is captured for the workflow_runs audit trail. func (s *Store) executeWorkflowActions(ctx context.Context, tenantID int64, doc *Document, actions []WorkflowAction) (json.RawMessage, error) { type actionResult struct { StepOrder int `json:"step_order"` ActionType string `json:"action_type"` OK bool `json:"ok"` Detail string `json:"detail,omitempty"` } results := make([]actionResult, 0, len(actions)) var firstErr error for _, a := range actions { res := actionResult{StepOrder: a.StepOrder, ActionType: a.ActionType} if err := s.executeWorkflowAction(ctx, tenantID, doc, a); err != nil { res.OK = false res.Detail = err.Error() if firstErr == nil { firstErr = err } } else { res.OK = true } results = append(results, res) } summary, err := json.Marshal(results) if err != nil { return nil, firstErr } return json.RawMessage(summary), firstErr } // executeWorkflowAction runs one workflow action against a document. func (s *Store) executeWorkflowAction(ctx context.Context, tenantID int64, doc *Document, a WorkflowAction) error { switch a.ActionType { case ActionAddTag: var c struct { TagID int64 `json:"tag_id"` } if err := json.Unmarshal(nonNilJSON(a.ActionConfig), &c); err != nil { return fmt.Errorf("add_tag config: %w", err) } // Ownership: only attach a tag that belongs to the tenant. var ok bool if err := s.db.QueryRow(ctx, `SELECT EXISTS(SELECT 1 FROM tags WHERE id = $1 AND tenant_id = $2)`, c.TagID, tenantID).Scan(&ok); err != nil { return fmt.Errorf("add_tag ownership check: %w", err) } if !ok { return fmt.Errorf("%w: tag_id %d", ErrTaxonomyNotFound, c.TagID) } return s.AttachTag(ctx, doc.ID, c.TagID) case ActionSetDocType: var c struct { DocTypeID int64 `json:"doc_type_id"` } if err := json.Unmarshal(nonNilJSON(a.ActionConfig), &c); err != nil { return fmt.Errorf("set_doc_type config: %w", err) } var ok bool if err := s.db.QueryRow(ctx, `SELECT EXISTS(SELECT 1 FROM document_types WHERE id = $1 AND tenant_id = $2)`, c.DocTypeID, tenantID).Scan(&ok); err != nil { return fmt.Errorf("set_doc_type ownership check: %w", err) } if !ok { return fmt.Errorf("%w: doc_type_id %d", ErrTaxonomyNotFound, c.DocTypeID) } if err := s.SetDocumentDocType(ctx, doc.ID, tenantID, c.DocTypeID); err != nil { return err } doc.DocTypeID = &c.DocTypeID return nil case ActionSetCorrespondent: var c struct { CorrespondentID int64 `json:"correspondent_id"` } if err := json.Unmarshal(nonNilJSON(a.ActionConfig), &c); err != nil { return fmt.Errorf("set_correspondent config: %w", err) } var ok bool if err := s.db.QueryRow(ctx, `SELECT EXISTS(SELECT 1 FROM correspondents WHERE id = $1 AND tenant_id = $2)`, c.CorrespondentID, tenantID).Scan(&ok); err != nil { return fmt.Errorf("set_correspondent ownership check: %w", err) } if !ok { return fmt.Errorf("%w: correspondent_id %d", ErrTaxonomyNotFound, c.CorrespondentID) } if err := s.SetDocumentCorrespondent(ctx, doc.ID, tenantID, c.CorrespondentID); err != nil { return err } doc.CorrespondentID = &c.CorrespondentID return nil case ActionApplyClassificationTmpl: var c struct { TemplateID int64 `json:"template_id"` Overwrite bool `json:"overwrite"` } if err := json.Unmarshal(nonNilJSON(a.ActionConfig), &c); err != nil { return fmt.Errorf("apply_classification_template config: %w", err) } _, err := s.ApplyTemplate(ctx, doc.ID, c.TemplateID, tenantID, c.Overwrite) return err case ActionSetCustomField: return s.applyWorkflowCustomField(ctx, tenantID, doc, a.ActionConfig) default: return fmt.Errorf("%w: unknown action_type %q", ErrInvalidWorkflowAction, a.ActionType) } } // applyWorkflowCustomField sets a single custom-field value on the document, // merging with the document's existing values (SetDocumentFieldValues is // full-replace, so overlay the one field over the current set) and validating // the value against the field type. func (s *Store) applyWorkflowCustomField(ctx context.Context, tenantID int64, doc *Document, cfg json.RawMessage) error { var c struct { FieldID int64 `json:"field_id"` Value *string `json:"value"` ValueText *string `json:"value_text"` ValueNumber *float64 `json:"value_number"` ValueDate *string `json:"value_date"` ValueBool *bool `json:"value_bool"` } if err := json.Unmarshal(nonNilJSON(cfg), &c); err != nil { return fmt.Errorf("set_custom_field config: %w", err) } if c.FieldID == 0 { return fmt.Errorf("%w: set_custom_field requires field_id", ErrInvalidWorkflowAction) } // Resolve the field type so a bare "value" string is routed to the correct // typed column. defs, err := s.ListCustomFieldDefs(ctx, tenantID) if err != nil { return err } var def *CustomFieldDef for i := range defs { if defs[i].ID == c.FieldID { def = &defs[i] break } } if def == nil { return fmt.Errorf("%w: field_id %d", ErrCustomFieldNotFound, c.FieldID) } newInput := DocumentFieldValueInput{FieldID: c.FieldID} // Explicit typed columns take precedence; otherwise route the bare "value". switch { case c.ValueText != nil || c.ValueNumber != nil || c.ValueDate != nil || c.ValueBool != nil: newInput.ValueText = c.ValueText newInput.ValueNumber = c.ValueNumber newInput.ValueDate = c.ValueDate newInput.ValueBool = c.ValueBool case c.Value != nil: switch def.FieldType { case "number", "monetary": f, perr := strconv.ParseFloat(*c.Value, 64) if perr != nil { return fmt.Errorf("set_custom_field: value %q not a number for field %q", *c.Value, def.Name) } newInput.ValueNumber = &f case "boolean": b, perr := strconv.ParseBool(*c.Value) if perr != nil { return fmt.Errorf("set_custom_field: value %q not a boolean for field %q", *c.Value, def.Name) } newInput.ValueBool = &b case "date": newInput.ValueDate = c.Value default: newInput.ValueText = c.Value } } // Merge with existing values (full-replace semantics of the setter). existing, err := s.ListDocumentFieldValues(ctx, doc.ID, tenantID) if err != nil { return err } inputs := make([]DocumentFieldValueInput, 0, len(existing)+1) replaced := false for _, v := range existing { if v.FieldID == c.FieldID { inputs = append(inputs, newInput) replaced = true continue } inputs = append(inputs, existingValueToInput(v)) } if !replaced { inputs = append(inputs, newInput) } if _, err := s.SetDocumentFieldValues(ctx, doc.ID, tenantID, inputs); err != nil { return err } return nil } // recordWorkflowRun inserts a workflow_runs audit row. Best-effort: a failure // to record is logged via the returned error being ignored by the caller // (the run's effects are already applied). errStr is stored as NULL when empty. func (s *Store) recordWorkflowRun(ctx context.Context, workflowID, tenantID int64, documentID *int64, matched bool, actionsApplied json.RawMessage, errStr string) { var applied []byte if len(actionsApplied) > 0 { applied = []byte(actionsApplied) } _, _ = s.db.Exec(ctx, ` INSERT INTO workflow_runs (workflow_id, tenant_id, document_id, matched, actions_applied, error) VALUES ($1, $2, $3, $4, $5, $6) `, workflowID, tenantID, documentID, matched, applied, nullIfEmpty(errStr)) } // ListWorkflowRuns returns the recorded runs of a workflow (most recent first), // scoped to tenant ownership of the workflow. Returns a non-nil slice. func (s *Store) ListWorkflowRuns(ctx context.Context, workflowID, tenantID int64, limit int) ([]WorkflowRun, error) { owns, err := s.workflowOwned(ctx, workflowID, tenantID) if err != nil { return nil, err } if !owns { return nil, ErrWorkflowNotFound } if limit <= 0 || limit > 500 { limit = 100 } rows, err := s.db.Query(ctx, ` SELECT id, workflow_id, tenant_id, document_id, triggered_at, matched, actions_applied, COALESCE(error, '') FROM workflow_runs WHERE workflow_id = $1 AND tenant_id = $2 ORDER BY triggered_at DESC, id DESC LIMIT $3`, workflowID, tenantID, limit) if err != nil { return nil, fmt.Errorf("storage: list workflow runs: %w", err) } defer rows.Close() out := make([]WorkflowRun, 0) for rows.Next() { var run WorkflowRun var applied []byte if err := rows.Scan(&run.ID, &run.WorkflowID, &run.TenantID, &run.DocumentID, &run.TriggeredAt, &run.Matched, &applied, &run.Error); err != nil { return nil, fmt.Errorf("storage: scan workflow run: %w", err) } if len(applied) > 0 { run.ActionsApplied = json.RawMessage(applied) } out = append(out, run) } return out, rows.Err() } // TestWorkflow performs a pure dry-run of a workflow: it evaluates the // condition tree against either an existing document (documentID set) or a // synthetic in-memory document built from rawText (documentID nil), reporting // which leaves matched and which actions WOULD run — never executing them. func (s *Store) TestWorkflow(ctx context.Context, workflowID, tenantID int64, documentID *int64, rawText string) (*WorkflowTestResult, error) { w, err := s.GetWorkflow(ctx, workflowID, tenantID) if err != nil { return nil, err } var doc *Document if documentID != nil { doc, err = s.GetDocument(ctx, *documentID, tenantID) if err != nil { if errors.Is(err, pgx.ErrNoRows) { return nil, ErrDocumentNotFound } return nil, err } } else { // Synthetic document: rawText feeds both ocr_text and title so text // leaves can be exercised without a real upload. doc = &Document{TenantID: tenantID, Title: rawText, OCRText: rawText, Source: "test"} } leaves := make([]LeafEvalResult, 0) matched, err := s.evalNode(ctx, w.ConditionTree, doc, &leaves) if err != nil { return nil, err } res := &WorkflowTestResult{ WorkflowID: workflowID, Matched: matched, Leaves: leaves, ActionsToRun: make([]WorkflowAction, 0), } if matched { res.ActionsToRun = w.Actions } return res, nil }