WeiCHEN
dca33c705d
Move bizdev/freedoc/intent/projectworkflow/squad giant files into ≤800-line use-case siblings without behavior change; short tests green.
109 lines
3.4 KiB
Go
109 lines
3.4 KiB
Go
package projectworkflow
|
|
|
|
import (
|
|
"context"
|
|
"encoding/json"
|
|
"errors"
|
|
"fmt"
|
|
"time"
|
|
|
|
"github.com/jackc/pgx/v5"
|
|
)
|
|
|
|
func (s *pgStore) CreateWorkflowRun(ctx context.Context, run *WorkflowRun) (*WorkflowRun, error) {
|
|
manifestJSON, _ := json.Marshal(run.InputManifest)
|
|
query := `
|
|
INSERT INTO workflow_runs (
|
|
run_id, project_id, workflow_key, status, started_by, stage_key, task_key,
|
|
input_manifest, output_doc_id, output_summary, error_message, started_at, created_at
|
|
) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, now())
|
|
RETURNING id, created_at
|
|
`
|
|
now := time.Now()
|
|
run.StartedAt = &now
|
|
err := s.pool.QueryRow(
|
|
ctx, query,
|
|
run.RunID, run.ProjectID, run.WorkflowKey, run.Status, run.StartedBy,
|
|
run.StageKey, run.TaskKey, manifestJSON, run.OutputDocID, run.OutputSummary,
|
|
run.ErrorMessage, run.StartedAt,
|
|
).Scan(&run.ID, &run.CreatedAt)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("create workflow run failed: %w", err)
|
|
}
|
|
return run, nil
|
|
}
|
|
|
|
func (s *pgStore) UpdateWorkflowRun(ctx context.Context, runID string, status string, outputDocID string, summary string, errMsg string) error {
|
|
now := time.Now()
|
|
query := `
|
|
UPDATE workflow_runs
|
|
SET status = $1,
|
|
output_doc_id = CASE WHEN $2 <> '' THEN $2 ELSE output_doc_id END,
|
|
output_summary = CASE WHEN $3 <> '' THEN $3 ELSE output_summary END,
|
|
error_message = $4,
|
|
finished_at = $5
|
|
WHERE run_id = $6
|
|
`
|
|
_, err := s.pool.Exec(ctx, query, status, outputDocID, summary, errMsg, now, runID)
|
|
return err
|
|
}
|
|
|
|
func (s *pgStore) GetWorkflowRun(ctx context.Context, runID string) (*WorkflowRun, error) {
|
|
query := `
|
|
SELECT id, run_id, project_id, workflow_key, status, started_by, stage_key, task_key,
|
|
input_manifest, output_doc_id, output_summary, error_message, started_at, finished_at, created_at
|
|
FROM workflow_runs
|
|
WHERE run_id = $1
|
|
`
|
|
run := &WorkflowRun{}
|
|
var manifestRaw []byte
|
|
err := s.pool.QueryRow(ctx, query, runID).Scan(
|
|
&run.ID, &run.RunID, &run.ProjectID, &run.WorkflowKey, &run.Status, &run.StartedBy,
|
|
&run.StageKey, &run.TaskKey, &manifestRaw, &run.OutputDocID, &run.OutputSummary,
|
|
&run.ErrorMessage, &run.StartedAt, &run.FinishedAt, &run.CreatedAt,
|
|
)
|
|
if errors.Is(err, pgx.ErrNoRows) {
|
|
return nil, errors.New("workflow run not found")
|
|
}
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
if len(manifestRaw) > 0 {
|
|
_ = json.Unmarshal(manifestRaw, &run.InputManifest)
|
|
}
|
|
return run, nil
|
|
}
|
|
|
|
func (s *pgStore) ListWorkflowRuns(ctx context.Context, projectID int64, workflowKey string, limit int) ([]*WorkflowRun, error) {
|
|
if limit <= 0 {
|
|
limit = 20
|
|
}
|
|
query := `
|
|
SELECT id, run_id, project_id, workflow_key, status, started_by, stage_key, task_key,
|
|
output_doc_id, output_summary, error_message, started_at, finished_at, created_at
|
|
FROM workflow_runs
|
|
WHERE project_id = $1 AND ($2 = '' OR workflow_key = $2)
|
|
ORDER BY id DESC
|
|
LIMIT $3
|
|
`
|
|
rows, err := s.pool.Query(ctx, query, projectID, workflowKey, limit)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
defer rows.Close()
|
|
|
|
var list []*WorkflowRun
|
|
for rows.Next() {
|
|
run := &WorkflowRun{}
|
|
if err := rows.Scan(
|
|
&run.ID, &run.RunID, &run.ProjectID, &run.WorkflowKey, &run.Status, &run.StartedBy,
|
|
&run.StageKey, &run.TaskKey, &run.OutputDocID, &run.OutputSummary,
|
|
&run.ErrorMessage, &run.StartedAt, &run.FinishedAt, &run.CreatedAt,
|
|
); err != nil {
|
|
return nil, err
|
|
}
|
|
list = append(list, run)
|
|
}
|
|
return list, rows.Err()
|
|
}
|