You've already forked agentic-coding-workflow
feat(jobs/runners): attachment cleanup + chat DeleteAttachments query/method
This commit is contained in:
@@ -30,3 +30,6 @@ JOIN messages m ON m.id = a.message_id
|
|||||||
JOIN conversations c ON c.id = m.conversation_id
|
JOIN conversations c ON c.id = m.conversation_id
|
||||||
WHERE c.deleted_at IS NOT NULL
|
WHERE c.deleted_at IS NOT NULL
|
||||||
AND c.deleted_at < NOW() - ($1::interval);
|
AND c.deleted_at < NOW() - ($1::interval);
|
||||||
|
|
||||||
|
-- name: DeleteAttachmentsByIDs :exec
|
||||||
|
DELETE FROM message_attachments WHERE id = ANY($1::uuid[]);
|
||||||
|
|||||||
@@ -58,6 +58,7 @@ type Repository interface {
|
|||||||
ListExpiredOrphans(ctx context.Context, olderThan time.Duration) ([]AttachmentRef, error)
|
ListExpiredOrphans(ctx context.Context, olderThan time.Duration) ([]AttachmentRef, error)
|
||||||
ListAttachmentsForSoftDeletedConversations(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
|
DeleteAttachment(ctx context.Context, id uuid.UUID) error
|
||||||
|
DeleteAttachments(ctx context.Context, ids []uuid.UUID) error
|
||||||
|
|
||||||
// ===== PromptTemplate =====
|
// ===== PromptTemplate =====
|
||||||
InsertPromptTemplate(ctx context.Context, p InsertTemplateParams) (*PromptTemplate, error)
|
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
|
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 =====
|
// ===== PromptTemplate =====
|
||||||
|
|
||||||
func (r *pgRepo) InsertPromptTemplate(ctx context.Context, p InsertTemplateParams) (*PromptTemplate, error) {
|
func (r *pgRepo) InsertPromptTemplate(ctx context.Context, p InsertTemplateParams) (*PromptTemplate, error) {
|
||||||
|
|||||||
@@ -244,6 +244,9 @@ func (f *fakeRepo) ListAttachmentsForSoftDeletedConversations(_ context.Context,
|
|||||||
panic("not implemented")
|
panic("not implemented")
|
||||||
}
|
}
|
||||||
func (f *fakeRepo) DeleteAttachment(_ context.Context, _ uuid.UUID) error { 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) {
|
func (f *fakeRepo) InsertPromptTemplate(_ context.Context, _ InsertTemplateParams) (*PromptTemplate, error) {
|
||||||
panic("not implemented")
|
panic("not implemented")
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -34,6 +34,15 @@ func (q *Queries) DeleteAttachment(ctx context.Context, id pgtype.UUID) error {
|
|||||||
return err
|
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
|
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
|
SELECT id, user_id, message_id, filename, mime_type, size_bytes, sha256, storage_path, created_at FROM message_attachments WHERE id = $1
|
||||||
`
|
`
|
||||||
|
|||||||
@@ -207,6 +207,9 @@ func (panicRepo) ListAttachmentsForSoftDeletedConversations(_ context.Context, _
|
|||||||
panic("not implemented")
|
panic("not implemented")
|
||||||
}
|
}
|
||||||
func (panicRepo) DeleteAttachment(_ context.Context, _ uuid.UUID) error { 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) {
|
func (panicRepo) InsertPromptTemplate(_ context.Context, _ InsertTemplateParams) (*PromptTemplate, error) {
|
||||||
panic("not implemented")
|
panic("not implemented")
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -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())
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -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")
|
||||||
|
}
|
||||||
Reference in New Issue
Block a user