package services import ( "context" "fmt" "reflect" "strings" "git.pengzhan.dev/haopengzhan/rsspipe/internal/db" "git.pengzhan.dev/haopengzhan/rsspipe/internal/events" "git.pengzhan.dev/haopengzhan/rsspipe/internal/logger" "git.pengzhan.dev/haopengzhan/rsspipe/internal/models" "go.uber.org/zap" ) type EntryService struct { db *db.DB bus *events.EventBus log *zap.Logger } func NewEntryService(database *db.DB, bus *events.EventBus) *EntryService { return &EntryService{ db: database, bus: bus, log: logger.Named("entry-service"), } } // IngestEntries handles bulk ingestion of parsed feed entries (from crawler or OPML import) func (s *EntryService) IngestEntries(ctx context.Context, feedID string, entries []models.FeedEntry, feedTitle, feedSiteURL string) ([]models.FeedEntry, error) { newEntries, err := s.db.UpsertFeedEntries(feedID, entries) if err != nil { return nil, err } for _, entry := range newEntries { entry.FeedTitle = feedTitle entry.FeedSiteURL = feedSiteURL diff := &models.EntryDiff{ FieldsChanged: []string{"id", "title", "content", "url"}, } if s.bus != nil { s.bus.Publish(ctx, events.Event{ Type: events.EventPostCreated, Entry: entry, NewEntry: &entry, Diff: diff, Depth: 0, }) s.bus.Publish(ctx, events.Event{ Type: events.EventEntryTransition, Entry: entry, NewEntry: &entry, Diff: diff, Depth: 0, }) } } return newEntries, nil } // PatchEntry applies atomic mutations to an entry, calculates the EntryDiff between Old and New states, // and fires EventEntryTransition to trigger downstream transition-based automation rules. func (s *EntryService) PatchEntry(ctx context.Context, entryID string, patch models.PatchEntryRequest) (*models.FeedEntry, *models.EntryDiff, error) { oldEntry, err := s.db.GetEntry(entryID) if err != nil { return nil, nil, fmt.Errorf("entry not found: %w", err) } diff := &models.EntryDiff{ FieldsChanged: make([]string, 0), VariantsAdded: make([]string, 0), VariantsChanged: make([]string, 0), VariantsRemoved: make([]string, 0), } // 1. Update IsRead if patch.IsRead != nil && *patch.IsRead != oldEntry.IsRead { if err := s.db.UpdateEntryRead(entryID, *patch.IsRead); err != nil { return nil, nil, err } diff.FieldsChanged = append(diff.FieldsChanged, "is_read") } // 2. Update IsStarred if patch.IsStarred != nil && *patch.IsStarred != oldEntry.IsStarred { if err := s.db.UpdateEntryStarred(entryID, *patch.IsStarred); err != nil { return nil, nil, err } diff.FieldsChanged = append(diff.FieldsChanged, "is_starred") } // 3. Update Metadata (e.g. tags) if patch.Metadata != nil { if oldEntry.Metadata == nil { oldEntry.Metadata = make(map[string]interface{}) } metadataChanged := false for k, v := range patch.Metadata { if !reflect.DeepEqual(oldEntry.Metadata[k], v) { oldEntry.Metadata[k] = v metadataChanged = true diff.FieldsChanged = append(diff.FieldsChanged, "metadata."+k) } } if metadataChanged { if err := s.db.UpdateEntryMetadata(entryID, oldEntry.Metadata); err != nil { return nil, nil, err } } } // 4. Update Variants if patch.Variants != nil { for vKey, vVal := range patch.Variants { if vVal == nil { // Variant removal if oldEntry.Variants != nil { if _, exists := oldEntry.Variants[vKey]; exists { _ = s.db.DeleteVariant(entryID, vKey) diff.VariantsRemoved = append(diff.VariantsRemoved, vKey) } } continue } // Variant add or update vVal.EntryID = entryID vVal.VariantKey = vKey oldVariant, exists := oldEntry.Variants[vKey] if !exists { diff.VariantsAdded = append(diff.VariantsAdded, vKey) } else if oldVariant.TextValue != vVal.TextValue || oldVariant.BlobPath != vVal.BlobPath || !reflect.DeepEqual(oldVariant.Metadata, vVal.Metadata) { diff.VariantsChanged = append(diff.VariantsChanged, vKey) } if err := s.db.SetVariant(*vVal); err != nil { return nil, nil, fmt.Errorf("failed to save variant %s: %w", vKey, err) } } } // Fetch fresh NewState newEntry, err := s.db.GetEntry(entryID) if err != nil { return nil, nil, fmt.Errorf("failed to retrieve updated entry: %w", err) } s.log.Debug("Entry patched with transition diff", zap.String("entry_id", entryID), zap.Strings("fields_changed", diff.FieldsChanged), zap.Strings("variants_added", diff.VariantsAdded), zap.Strings("variants_changed", diff.VariantsChanged), ) // Publish transition event to trigger reactive rules if s.bus != nil { s.bus.Publish(ctx, events.Event{ Type: events.EventEntryTransition, Entry: *newEntry, OldEntry: oldEntry, NewEntry: newEntry, Diff: diff, Depth: 0, }) s.bus.Publish(ctx, events.Event{ Type: events.EventPostUpdated, Entry: *newEntry, OldEntry: oldEntry, NewEntry: newEntry, Diff: diff, Depth: 0, }) for _, vKey := range diff.VariantsAdded { s.bus.Publish(ctx, events.Event{ Type: events.EventVariantUpdated, Entry: *newEntry, VariantKey: vKey, Depth: 0, }) } for _, vKey := range diff.VariantsChanged { s.bus.Publish(ctx, events.Event{ Type: events.EventVariantUpdated, Entry: *newEntry, VariantKey: vKey, Depth: 0, }) } } return newEntry, diff, nil } func (s *EntryService) GetEntry(ctx context.Context, id string) (*models.FeedEntry, error) { return s.db.GetEntry(id) } func (s *EntryService) ListEntries(ctx context.Context, filter db.EntryFilter) ([]models.FeedEntry, int, error) { return s.db.ListEntries(filter) } func (s *EntryService) MarkAllRead(ctx context.Context, feedID, categoryID *string) error { return s.db.MarkAllRead(feedID, categoryID) } func (s *EntryService) ToggleShare(ctx context.Context, id string) (*string, error) { return s.db.ToggleShareEntry(id) } func (s *EntryService) GetSharedEntry(ctx context.Context, shareUUID string) (*models.FeedEntry, error) { return s.db.GetSharedEntry(strings.TrimSuffix(shareUUID, ".html")) }