package filemanagerrealtime import ( "context" "errors" "log/slog" "strings" "time" "gorm.io/gorm" filemanager "wucher/internal/domain/file_manager" "wucher/internal/shared/pkg/uuidv7" ) type StatusPollerConfig struct { PollInterval time.Duration BatchSize int OnPublished func(context.Context, FileStatusEvent) error } type StatusPoller struct { db *gorm.DB hub *Hub logger *slog.Logger config StatusPollerConfig } func NewStatusPoller(db *gorm.DB, hub *Hub, logger *slog.Logger, cfg StatusPollerConfig) *StatusPoller { if cfg.PollInterval <= 0 { cfg.PollInterval = time.Second } if cfg.BatchSize <= 0 { cfg.BatchSize = 100 } if logger == nil { logger = slog.Default() } return &StatusPoller{ db: db, hub: hub, logger: logger.With("component", "file-status-poller"), config: cfg, } } func (p *StatusPoller) Run(ctx context.Context) error { if p == nil { return nil } if p.db == nil { return errors.New("status poller requires database") } if p.hub == nil { return errors.New("status poller requires hub") } ticker := time.NewTicker(p.config.PollInterval) defer ticker.Stop() for { processed, err := p.PollOnce(ctx) if err != nil { if ctx.Err() != nil { return nil } p.logger.Error("file status poll failed", slog.Any("error", err)) } if ctx.Err() != nil { return nil } if processed > 0 { continue } select { case <-ctx.Done(): return nil case <-ticker.C: } } } func (p *StatusPoller) PollOnce(ctx context.Context) (int, error) { if p == nil { return 0, nil } if p.db == nil { return 0, errors.New("status poller requires database") } if p.hub == nil { return 0, errors.New("status poller requires hub") } totalPublished := 0 for { rows, err := p.listPendingEvents(ctx) if err != nil { return totalPublished, err } if len(rows) == 0 { return totalPublished, nil } deliveredAt := time.Now().UTC() ids := make([][]byte, 0, len(rows)) metadataByFileID, metadataErr := p.loadFileMetadata(ctx, rows) if metadataErr != nil { p.logger.Warn("file status metadata enrich failed", slog.Any("error", metadataErr)) } for i := range rows { ids = append(ids, cloneBytes(rows[i].ID)) meta := metadataByFileID[string(rows[i].FileID)] event, recipient, ok := buildFileStatusEventFromRealtimeEvent(&rows[i], meta) if !ok { continue } if p.hub.PublishToUsers([]string{recipient}, event) > 0 { totalPublished++ } if p.config.OnPublished != nil { if err := p.config.OnPublished(ctx, event); err != nil { p.logger.Warn("file status post-publish callback failed", slog.Any("error", err), slog.String("file_uuid", event.FileUUID), slog.String("status", event.Status)) } } } if err := p.markDelivered(ctx, ids, deliveredAt); err != nil { return totalPublished, err } if len(rows) < p.config.BatchSize { return totalPublished, nil } } } type fileStatusEventFileMetadata struct { FolderUUID *string MimeType string } func (p *StatusPoller) loadFileMetadata(ctx context.Context, events []filemanager.FileStatusRealtimeEvent) (map[string]fileStatusEventFileMetadata, error) { result := make(map[string]fileStatusEventFileMetadata, len(events)) if p == nil || p.db == nil || len(events) == 0 { return result, nil } fileIDs := make([][]byte, 0, len(events)) seen := make(map[string]struct{}, len(events)) for i := range events { key := string(events[i].FileID) if key == "" { continue } if _, ok := seen[key]; ok { continue } seen[key] = struct{}{} fileIDs = append(fileIDs, cloneBytes(events[i].FileID)) } if len(fileIDs) == 0 { return result, nil } type fileRow struct { ID []byte `gorm:"column:id"` FolderID []byte `gorm:"column:folder_id"` MimeType string `gorm:"column:mime_type"` } rows := make([]fileRow, 0, len(fileIDs)) if err := p.db.WithContext(ctx). Model(&filemanager.File{}). Select("id, folder_id, mime_type"). Where("id IN ?", fileIDs). Find(&rows).Error; err != nil { return result, err } for i := range rows { meta := fileStatusEventFileMetadata{ MimeType: strings.TrimSpace(rows[i].MimeType), } if len(rows[i].FolderID) == 16 { if folderUUID, err := uuidv7.BytesToString(rows[i].FolderID); err == nil { folderUUID = strings.ToLower(strings.TrimSpace(folderUUID)) if folderUUID != "" { meta.FolderUUID = &folderUUID } } } result[string(rows[i].ID)] = meta } return result, nil } func (p *StatusPoller) listPendingEvents(ctx context.Context) ([]filemanager.FileStatusRealtimeEvent, error) { rows := make([]filemanager.FileStatusRealtimeEvent, 0, p.config.BatchSize) err := p.db.WithContext(ctx). Model(&filemanager.FileStatusRealtimeEvent{}). Where("delivered_at IS NULL"). Order("occurred_at ASC"). Order("id ASC"). Limit(p.config.BatchSize). Find(&rows).Error return rows, err } func (p *StatusPoller) markDelivered(ctx context.Context, ids [][]byte, deliveredAt time.Time) error { if len(ids) == 0 { return nil } deliveredAt = deliveredAt.UTC() return p.db.WithContext(ctx). Model(&filemanager.FileStatusRealtimeEvent{}). Where("id IN ?", ids). Where("delivered_at IS NULL"). Updates(map[string]any{ "delivered_at": deliveredAt, "updated_at": deliveredAt, }).Error } func buildFileStatusEventFromRealtimeEvent(row *filemanager.FileStatusRealtimeEvent, meta fileStatusEventFileMetadata) (FileStatusEvent, string, bool) { if row == nil { return FileStatusEvent{}, "", false } status := filemanager.NormalizeFileStatus(row.Status) if !isRealtimeStatus(status) { return FileStatusEvent{}, "", false } fileUUID, err := uuidv7.BytesToString(row.FileID) if err != nil || strings.TrimSpace(fileUUID) == "" { return FileStatusEvent{}, "", false } userUUID, err := uuidv7.BytesToString(row.UserID) if err != nil { return FileStatusEvent{}, "", false } userUUID = strings.ToLower(strings.TrimSpace(userUUID)) if userUUID == "" { return FileStatusEvent{}, "", false } reason := strings.TrimSpace(row.FailureReason) var failureReason *string if reason != "" { failureReason = &reason } updatedAt := row.OccurredAt.UTC() if updatedAt.IsZero() { updatedAt = row.UpdatedAt.UTC() } if updatedAt.IsZero() { updatedAt = time.Now().UTC() } return FileStatusEvent{ EventType: FileStatusChangedEventType, FileUUID: fileUUID, FolderUUID: meta.FolderUUID, Status: status, Name: row.Name, MimeType: meta.MimeType, FailureReason: failureReason, UpdatedAt: updatedAt.Format(time.RFC3339), }, userUUID, true } func isRealtimeStatus(status string) bool { switch filemanager.NormalizeFileStatus(status) { case filemanager.FileStatusUploaded, filemanager.FileStatusProcessing, filemanager.FileStatusValidated, filemanager.FileStatusFailed, filemanager.FileStatusReady, filemanager.FileStatusTrashed: return true default: return false } } func cloneBytes(raw []byte) []byte { if len(raw) == 0 { return nil } return append([]byte(nil), raw...) }