From f8526d45492265ce63431ccebef26620a6bbccdf Mon Sep 17 00:00:00 2001 From: Mathias Date: Fri, 2 Oct 2026 08:51:39 +0200 Subject: [PATCH] feat(store): SaveProgress and Heartbeat; Get returns the phases and both timestamps (ADR-035) Co-Authored-By: Claude Sonnet 5.5 --- .../adapters/store/uploaded_transcripts.go | 41 +++++++++++++++++-- 1 file changed, 37 insertions(+), 4 deletions(-) diff --git a/internal/adapters/store/uploaded_transcripts.go b/internal/adapters/store/uploaded_transcripts.go index c5820fe..71d289b 100644 --- a/internal/adapters/store/uploaded_transcripts.go +++ b/internal/adapters/store/uploaded_transcripts.go @@ -68,6 +68,26 @@ func (s *Store) CompleteUpload(ctx context.Context, userID, id string, t domain. t.Text, segs, t.Endpoint, t.AudioSeconds, rec) } +// SaveProgress records the job's phases and when something last really changed +// (ADR-035). It replaces the previous snapshot: the Tracker holds the whole plan. +func (s *Store) SaveProgress(ctx context.Context, userID, id string, steps []domain.ProgressStep, at time.Time) error { + b, err := json.Marshal(steps) + if err != nil { + return fmt.Errorf("store: save progress: marshal: %w", err) + } + return s.updateUpload(ctx, userID, id, "save progress", + `UPDATE uploaded_transcripts SET progress = $3, progress_at = $4 + WHERE id = $1 AND user_id = $2`, b, at) +} + +// Heartbeat stamps the upload as alive now. It does not touch progress_at: alive says +// the service is still reporting, not that work is moving (ADR-035). +func (s *Store) Heartbeat(ctx context.Context, userID, id string) error { + return s.updateUpload(ctx, userID, id, "heartbeat", + `UPDATE uploaded_transcripts SET alive_at = NOW() + WHERE id = $1 AND user_id = $2`) +} + // FailUpload moves an upload to failed with a typed reason. func (s *Store) FailUpload(ctx context.Context, userID, id string, kind domain.UploadErrorKind) error { return s.updateUpload(ctx, userID, id, "fail upload", @@ -165,19 +185,21 @@ func (s *Store) FailInterruptedUploads(ctx context.Context) (int, error) { } const uploadSelect = `SELECT id, user_id, title, status, content, segments, endpoint, - audio_seconds, error_kind, had_video, reconciliation, created_at, finished_at FROM uploaded_transcripts` + audio_seconds, error_kind, had_video, reconciliation, progress, progress_at, alive_at, + created_at, finished_at FROM uploaded_transcripts` // scanUpload scans one uploadSelect row. withText controls whether the transcript // body and segments are kept (GetUpload) or dropped (ListUploads). func scanUpload(row pgx.Row, u *domain.UploadedTranscript, withText bool) error { var ( status, errKind, endpoint, content *string - segments, reconciliation []byte + segments, reconciliation, progress []byte seconds *float64 - finished *time.Time + finished, progressAt, aliveAt *time.Time ) if err := row.Scan(&u.ID, &u.UserID, &u.Title, &status, &content, &segments, &endpoint, - &seconds, &errKind, &u.HadVideo, &reconciliation, &u.CreatedAt, &finished); err != nil { + &seconds, &errKind, &u.HadVideo, &reconciliation, &progress, &progressAt, &aliveAt, + &u.CreatedAt, &finished); err != nil { return err } u.Status = domain.UploadStatus(deref(status)) @@ -192,6 +214,17 @@ func scanUpload(row pgx.Row, u *domain.UploadedTranscript, withText bool) error if !withText { return nil } + if len(progress) > 0 { + if err := json.Unmarshal(progress, &u.Progress); err != nil { + return fmt.Errorf("unmarshal progress: %w", err) + } + } + if progressAt != nil { + u.ProgressAt = *progressAt + } + if aliveAt != nil { + u.AliveAt = *aliveAt + } if len(reconciliation) > 0 { var r domain.Reconciliation if err := json.Unmarshal(reconciliation, &r); err != nil {