You've already forked agentic-coding-workflow
88 lines
2.8 KiB
Go
88 lines
2.8 KiB
Go
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())
|
|
}
|
|
}
|