From f56fe449fdcc12ab82228894f6e4b88df2adc768 Mon Sep 17 00:00:00 2001 From: Jerry Yan <792602257@qq.com> Date: Wed, 6 May 2026 08:07:49 +0800 Subject: [PATCH] feat(jobs/runners): attachment cleanup + chat DeleteAttachments query/method --- internal/chat/queries/attachments.sql | 3 + internal/chat/repository.go | 15 ++++ internal/chat/service_endpoint_test.go | 3 + internal/chat/sqlc/attachments.sql.go | 9 ++ internal/chat/streamer_test.go | 3 + internal/jobs/runners/attachment_cleanup.go | 87 +++++++++++++++++++ .../jobs/runners/attachment_cleanup_test.go | 73 ++++++++++++++++ 7 files changed, 193 insertions(+) create mode 100644 internal/jobs/runners/attachment_cleanup.go create mode 100644 internal/jobs/runners/attachment_cleanup_test.go diff --git a/internal/chat/queries/attachments.sql b/internal/chat/queries/attachments.sql index 511b985..82c9755 100644 --- a/internal/chat/queries/attachments.sql +++ b/internal/chat/queries/attachments.sql @@ -30,3 +30,6 @@ JOIN messages m ON m.id = a.message_id JOIN conversations c ON c.id = m.conversation_id WHERE c.deleted_at IS NOT NULL AND c.deleted_at < NOW() - ($1::interval); + +-- name: DeleteAttachmentsByIDs :exec +DELETE FROM message_attachments WHERE id = ANY($1::uuid[]); diff --git a/internal/chat/repository.go b/internal/chat/repository.go index 056304a..793158f 100644 --- a/internal/chat/repository.go +++ b/internal/chat/repository.go @@ -58,6 +58,7 @@ type Repository interface { ListExpiredOrphans(ctx context.Context, olderThan time.Duration) ([]AttachmentRef, error) ListAttachmentsForSoftDeletedConversations(ctx context.Context, olderThan time.Duration) ([]AttachmentRef, error) DeleteAttachment(ctx context.Context, id uuid.UUID) error + DeleteAttachments(ctx context.Context, ids []uuid.UUID) error // ===== PromptTemplate ===== InsertPromptTemplate(ctx context.Context, p InsertTemplateParams) (*PromptTemplate, error) @@ -778,6 +779,20 @@ func (r *pgRepo) DeleteAttachment(ctx context.Context, id uuid.UUID) error { return nil } +func (r *pgRepo) DeleteAttachments(ctx context.Context, ids []uuid.UUID) error { + if len(ids) == 0 { + return nil + } + pgIDs := make([]pgtype.UUID, len(ids)) + for i, id := range ids { + pgIDs[i] = toPgUUID(id) + } + if err := r.q.DeleteAttachmentsByIDs(ctx, pgIDs); err != nil { + return errs.Wrap(err, errs.CodeInternal, "delete attachments") + } + return nil +} + // ===== PromptTemplate ===== func (r *pgRepo) InsertPromptTemplate(ctx context.Context, p InsertTemplateParams) (*PromptTemplate, error) { diff --git a/internal/chat/service_endpoint_test.go b/internal/chat/service_endpoint_test.go index 7a5fb4d..3fc959c 100644 --- a/internal/chat/service_endpoint_test.go +++ b/internal/chat/service_endpoint_test.go @@ -244,6 +244,9 @@ func (f *fakeRepo) ListAttachmentsForSoftDeletedConversations(_ context.Context, panic("not implemented") } func (f *fakeRepo) DeleteAttachment(_ context.Context, _ uuid.UUID) error { panic("not implemented") } +func (f *fakeRepo) DeleteAttachments(_ context.Context, _ []uuid.UUID) error { + panic("not implemented") +} func (f *fakeRepo) InsertPromptTemplate(_ context.Context, _ InsertTemplateParams) (*PromptTemplate, error) { panic("not implemented") } diff --git a/internal/chat/sqlc/attachments.sql.go b/internal/chat/sqlc/attachments.sql.go index ba2df1f..02af113 100644 --- a/internal/chat/sqlc/attachments.sql.go +++ b/internal/chat/sqlc/attachments.sql.go @@ -34,6 +34,15 @@ func (q *Queries) DeleteAttachment(ctx context.Context, id pgtype.UUID) error { return err } +const deleteAttachmentsByIDs = `-- name: DeleteAttachmentsByIDs :exec +DELETE FROM message_attachments WHERE id = ANY($1::uuid[]) +` + +func (q *Queries) DeleteAttachmentsByIDs(ctx context.Context, dollar_1 []pgtype.UUID) error { + _, err := q.db.Exec(ctx, deleteAttachmentsByIDs, dollar_1) + return err +} + const getAttachment = `-- name: GetAttachment :one SELECT id, user_id, message_id, filename, mime_type, size_bytes, sha256, storage_path, created_at FROM message_attachments WHERE id = $1 ` diff --git a/internal/chat/streamer_test.go b/internal/chat/streamer_test.go index 1c78fe7..b1186e9 100644 --- a/internal/chat/streamer_test.go +++ b/internal/chat/streamer_test.go @@ -207,6 +207,9 @@ func (panicRepo) ListAttachmentsForSoftDeletedConversations(_ context.Context, _ panic("not implemented") } func (panicRepo) DeleteAttachment(_ context.Context, _ uuid.UUID) error { panic("not implemented") } +func (panicRepo) DeleteAttachments(_ context.Context, _ []uuid.UUID) error { + panic("not implemented") +} func (panicRepo) InsertPromptTemplate(_ context.Context, _ InsertTemplateParams) (*PromptTemplate, error) { panic("not implemented") } diff --git a/internal/jobs/runners/attachment_cleanup.go b/internal/jobs/runners/attachment_cleanup.go new file mode 100644 index 0000000..3d2fb54 --- /dev/null +++ b/internal/jobs/runners/attachment_cleanup.go @@ -0,0 +1,87 @@ +package runners + +import ( + "context" + "log/slog" + "time" + + "github.com/google/uuid" +) + +// AttachmentRow projects the chat.AttachmentRef fields the cleanup needs. The +// production adapter (T16) maps chat.AttachmentRef → AttachmentRow 1:1. +type AttachmentRow struct { + ID uuid.UUID + StoragePath string +} + +// AttachmentCleanupRepo is the minimal repo surface needed. olderThan matches +// chat.Repository's existing signature (time.Duration, not absolute cutoff). +type AttachmentCleanupRepo interface { + ListAttachmentsForSoftDeletedConversations(ctx context.Context, olderThan time.Duration) ([]AttachmentRow, error) + ListExpiredOrphans(ctx context.Context, olderThan time.Duration) ([]AttachmentRow, error) + DeleteAttachments(ctx context.Context, ids []uuid.UUID) error +} + +// AttachmentStorage is the storage.Storage surface for delete. +type AttachmentStorage interface { + Delete(ctx context.Context, storagePath string) error +} + +// AttachmentCleanupConfig matches jobs.AttachmentCleanupConfig minus enabled/interval. +type AttachmentCleanupConfig struct { + Retention time.Duration +} + +// AttachmentCleanup is the C runner: clears both soft-deleted-conversation +// attachments and orphan attachments older than Retention. +type AttachmentCleanup struct { + repo AttachmentCleanupRepo + st AttachmentStorage + cfg AttachmentCleanupConfig + log *slog.Logger +} + +// NewAttachmentCleanup constructs the runner. +func NewAttachmentCleanup(repo AttachmentCleanupRepo, st AttachmentStorage, cfg AttachmentCleanupConfig, log *slog.Logger) *AttachmentCleanup { + if log == nil { + log = slog.Default() + } + return &AttachmentCleanup{repo: repo, st: st, cfg: cfg, log: log} +} + +// Run runs both subphases. +func (r *AttachmentCleanup) Run(ctx context.Context) { + r.cleanup(ctx, "soft_deleted", func(ctx context.Context) ([]AttachmentRow, error) { + return r.repo.ListAttachmentsForSoftDeletedConversations(ctx, r.cfg.Retention) + }) + r.cleanup(ctx, "orphan", func(ctx context.Context) ([]AttachmentRow, error) { + return r.repo.ListExpiredOrphans(ctx, r.cfg.Retention) + }) +} + +func (r *AttachmentCleanup) cleanup(ctx context.Context, label string, list func(ctx context.Context) ([]AttachmentRow, error)) { + rows, err := list(ctx) + if err != nil { + r.log.Error("attachment_cleanup.list", "label", label, "err", err.Error()) + return + } + if len(rows) == 0 { + return + } + deletedIDs := make([]uuid.UUID, 0, len(rows)) + for _, row := range rows { + if err := r.st.Delete(ctx, row.StoragePath); err != nil { + r.log.Warn("attachment_cleanup.storage_delete_failed", + "label", label, "id", row.ID, "path", row.StoragePath, "err", err.Error()) + continue + } + deletedIDs = append(deletedIDs, row.ID) + } + if len(deletedIDs) == 0 { + return + } + if err := r.repo.DeleteAttachments(ctx, deletedIDs); err != nil { + r.log.Error("attachment_cleanup.db_delete", "label", label, "err", err.Error()) + } +} diff --git a/internal/jobs/runners/attachment_cleanup_test.go b/internal/jobs/runners/attachment_cleanup_test.go new file mode 100644 index 0000000..f6b758e --- /dev/null +++ b/internal/jobs/runners/attachment_cleanup_test.go @@ -0,0 +1,73 @@ +package runners_test + +import ( + "context" + "errors" + "testing" + "time" + + "github.com/google/uuid" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + "github.com/yan1h/agent-coding-workflow/internal/jobs/runners" +) + +type fakeAttachmentRepo struct { + softDel []runners.AttachmentRow + orphan []runners.AttachmentRow + deletedIDs []uuid.UUID +} + +func (f *fakeAttachmentRepo) ListAttachmentsForSoftDeletedConversations(_ context.Context, _ time.Duration) ([]runners.AttachmentRow, error) { + return f.softDel, nil +} +func (f *fakeAttachmentRepo) ListExpiredOrphans(_ context.Context, _ time.Duration) ([]runners.AttachmentRow, error) { + return f.orphan, nil +} +func (f *fakeAttachmentRepo) DeleteAttachments(_ context.Context, ids []uuid.UUID) error { + f.deletedIDs = append(f.deletedIDs, ids...) + return nil +} + +type fakeStorage struct { + deleted []string + err error +} + +func (s *fakeStorage) Delete(_ context.Context, path string) error { + if s.err != nil { + return s.err + } + s.deleted = append(s.deleted, path) + return nil +} + +func TestAttachmentCleanup_DeletesBothCategories(t *testing.T) { + t.Parallel() + a, b, c := uuid.New(), uuid.New(), uuid.New() + repo := &fakeAttachmentRepo{ + softDel: []runners.AttachmentRow{{ID: a, StoragePath: "a"}, {ID: b, StoragePath: "b"}}, + orphan: []runners.AttachmentRow{{ID: c, StoragePath: "c"}}, + } + st := &fakeStorage{} + r := runners.NewAttachmentCleanup(repo, st, runners.AttachmentCleanupConfig{Retention: 30 * 24 * time.Hour}, discardLogger()) + r.Run(context.Background()) + assert.ElementsMatch(t, []string{"a", "b", "c"}, st.deleted) + assert.ElementsMatch(t, []uuid.UUID{a, b, c}, repo.deletedIDs) +} + +func TestAttachmentCleanup_StorageErrorDoesNotDeleteFromDB(t *testing.T) { + t.Parallel() + repo := &fakeAttachmentRepo{ + softDel: []runners.AttachmentRow{ + {ID: uuid.New(), StoragePath: "fail"}, + {ID: uuid.New(), StoragePath: "ok"}, + }, + } + st := &fakeStorage{err: errors.New("disk full")} + r := runners.NewAttachmentCleanup(repo, st, runners.AttachmentCleanupConfig{Retention: 30 * 24 * time.Hour}, discardLogger()) + r.Run(context.Background()) + require.Empty(t, st.deleted, "storage error → no path recorded") + require.Empty(t, repo.deletedIDs, "storage error → no DB delete batch") +}