Full article extracted content...
", + } + + updated2, diff2, err := svc.PatchEntry(context.Background(), ingested[0].ID, models.PatchEntryRequest{ + Variants: map[string]*models.EntryVariant{ + "readability_full_text": fullTextVariant, + }, + }) + if err != nil { + t.Fatalf("failed to patch variant: %v", err) + } + + if !diff2.HasVariantAddition("readability_full_text") { + t.Errorf("expected diff2 to report variant addition of readability_full_text") + } + if updated2.Variants["readability_full_text"].TextValue == "" { + t.Errorf("expected variant to be loaded in updated entry") + } +} diff --git a/internal/services/feed_service.go b/internal/services/feed_service.go new file mode 100644 index 0000000..dc94cff --- /dev/null +++ b/internal/services/feed_service.go @@ -0,0 +1,186 @@ +package services + +import ( + "context" + "fmt" + "strings" + + "git.pengzhan.dev/haopengzhan/rsspipe/internal/crawler" + "git.pengzhan.dev/haopengzhan/rsspipe/internal/db" + "git.pengzhan.dev/haopengzhan/rsspipe/internal/logger" + "git.pengzhan.dev/haopengzhan/rsspipe/internal/models" + "git.pengzhan.dev/haopengzhan/rsspipe/internal/opml" + "git.pengzhan.dev/haopengzhan/rsspipe/internal/parser" + "go.uber.org/zap" +) + +type FeedService struct { + db *db.DB + entryService *EntryService + parser *parser.Parser + crawler *crawler.Crawler + log *zap.Logger +} + +func NewFeedService(database *db.DB, entrySvc *EntryService, cr *crawler.Crawler) *FeedService { + return &FeedService{ + db: database, + entryService: entrySvc, + parser: parser.NewParser(), + crawler: cr, + log: logger.Named("feed-service"), + } +} + +func (s *FeedService) ListFeeds(ctx context.Context) ([]models.Feed, error) { + return s.db.ListFeeds() +} + +func (s *FeedService) GetFeed(ctx context.Context, id string) (*models.Feed, error) { + return s.db.GetFeed(id) +} + +func (s *FeedService) CreateFeed(ctx context.Context, req models.CreateFeedRequest) (*models.Feed, error) { + feedURL := strings.TrimSpace(req.FeedURL) + if feedURL == "" { + return nil, fmt.Errorf("feed_url is required") + } + + if existing, _ := s.db.GetFeedByURL(feedURL); existing != nil { + return nil, fmt.Errorf("feed is already subscribed") + } + + res, err := s.parser.FetchFeed(ctx, feedURL, "", "") + if err != nil { + return nil, fmt.Errorf("failed to fetch feed: %w", err) + } + + title := res.Title + if req.Title != nil && strings.TrimSpace(*req.Title) != "" { + title = strings.TrimSpace(*req.Title) + } + iconURL := "" + if req.IconURL != nil && strings.TrimSpace(*req.IconURL) != "" { + iconURL = strings.TrimSpace(*req.IconURL) + } + + feed, err := s.db.CreateFeed(title, feedURL, res.SiteURL, iconURL, req.CategoryID) + if err != nil { + return nil, err + } + + if len(res.Entries) > 0 { + _, _ = s.entryService.IngestEntries(ctx, feed.ID, res.Entries, feed.Title, feed.SiteURL) + _ = s.db.UpdateFeedFetchStatus(feed.ID, res.ETag, res.LastModified, "") + } + + return s.db.GetFeed(feed.ID) +} + +func (s *FeedService) UpdateFeed(ctx context.Context, id string, req models.UpdateFeedRequest) (*models.Feed, error) { + if err := s.db.UpdateFeed(id, req); err != nil { + return nil, err + } + return s.db.GetFeed(id) +} + +func (s *FeedService) DeleteFeed(ctx context.Context, id string) error { + return s.db.DeleteFeed(id) +} + +func (s *FeedService) RefreshFeed(ctx context.Context, id string) (*models.Feed, error) { + if s.crawler != nil { + if err := s.crawler.RefreshSingle(ctx, id); err != nil { + return nil, err + } + } + return s.db.GetFeed(id) +} + +func (s *FeedService) RefreshAllFeeds(ctx context.Context) { + if s.crawler != nil { + go s.crawler.RefreshAll(context.Background()) + } +} + +// Category methods +func (s *FeedService) ListCategories(ctx context.Context) ([]models.Category, error) { + return s.db.ListCategories() +} + +func (s *FeedService) CreateCategory(ctx context.Context, title string, parentID *string) (*models.Category, error) { + return s.db.CreateCategory(strings.TrimSpace(title), parentID) +} + +func (s *FeedService) DeleteCategory(ctx context.Context, id string) error { + return s.db.DeleteCategory(id) +} + +func (s *FeedService) GetStats(ctx context.Context) (*models.Stats, error) { + return s.db.GetStats() +} + +// OPML methods +func (s *FeedService) ImportOPML(ctx context.Context, data []byte) (int, int, error) { + importedFeeds, err := opml.ParseOPML(data) + if err != nil { + return 0, 0, fmt.Errorf("invalid OPML XML: %w", err) + } + + categoryCache := make(map[string]string) + cats, _ := s.db.ListCategories() + for _, c := range cats { + categoryCache[c.Title] = c.ID + } + + createdCount := 0 + for _, item := range importedFeeds { + if item.FeedURL == "" { + continue + } + if existing, _ := s.db.GetFeedByURL(item.FeedURL); existing != nil { + continue + } + + var catID *string + if item.CategoryTitle != "" { + if cid, ok := categoryCache[item.CategoryTitle]; ok { + catID = &cid + } else { + newCat, err := s.db.CreateCategory(item.CategoryTitle, nil) + if err == nil { + categoryCache[item.CategoryTitle] = newCat.ID + catID = &newCat.ID + } + } + } + + title := item.Title + if title == "" { + title = item.FeedURL + } + + _, err := s.db.CreateFeed(title, item.FeedURL, item.SiteURL, "", catID) + if err == nil { + createdCount++ + } + } + + if s.crawler != nil { + go s.crawler.RefreshAll(context.Background()) + } + + return createdCount, len(importedFeeds), nil +} + +func (s *FeedService) ExportOPML(ctx context.Context) ([]byte, error) { + feeds, err := s.db.ListFeeds() + if err != nil { + return nil, err + } + cats, err := s.db.ListCategories() + if err != nil { + return nil, err + } + return opml.GenerateOPML(feeds, cats) +} diff --git a/internal/services/job_service.go b/internal/services/job_service.go new file mode 100644 index 0000000..862c82e --- /dev/null +++ b/internal/services/job_service.go @@ -0,0 +1,126 @@ +package services + +import ( + "context" + "crypto/sha256" + "encoding/hex" + "fmt" + "time" + + "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" + "github.com/google/uuid" + "go.uber.org/zap" +) + +type JobService struct { + db *db.DB + bus *events.EventBus + log *zap.Logger +} + +func NewJobService(database *db.DB, bus *events.EventBus) *JobService { + return &JobService{ + db: database, + bus: bus, + log: logger.Named("job-service"), + } +} + +// GenerateIdempotencyKey creates a deterministic key to prevent duplicate jobs +func GenerateIdempotencyKey(jobType, entityID, action string, extra string) string { + raw := fmt.Sprintf("%s:%s:%s:%s", jobType, entityID, action, extra) + h := sha256.Sum256([]byte(raw)) + return hex.EncodeToString(h[:16]) +} + +// Enqueue submits a job to the persistent SQLite queue. +// If an idempotency key is provided and a job already exists with that key, returns the existing job. +func (s *JobService) Enqueue(ctx context.Context, jobType string, payload map[string]interface{}, idempotencyKey string) (*models.Job, error) { + if idempotencyKey != "" { + if existing, err := s.db.GetJobByIdempotencyKey(idempotencyKey); err == nil && existing != nil { + s.log.Debug("Job already exists with idempotency key, skipping duplicate enqueue", + zap.String("job_id", existing.ID), + zap.String("type", jobType), + zap.String("key", idempotencyKey), + ) + return existing, nil + } + } + + job := models.Job{ + ID: uuid.New().String(), + Type: jobType, + Payload: payload, + Status: "pending", + Progress: 0, + Attempts: 0, + MaxAttempts: 3, + IdempotencyKey: idempotencyKey, + RunAt: time.Now().UTC(), + CreatedAt: time.Now().UTC(), + UpdatedAt: time.Now().UTC(), + } + + created, err := s.db.CreateJob(job) + if err != nil { + return nil, fmt.Errorf("failed to enqueue job: %w", err) + } + + s.log.Info("Job enqueued", + zap.String("job_id", created.ID), + zap.String("type", created.Type), + zap.String("idempotency_key", idempotencyKey), + ) + + if s.bus != nil { + s.bus.Publish(ctx, events.Event{ + Type: events.EventJobProgress, + Job: created, + }) + } + + return created, nil +} + +func (s *JobService) GetJob(ctx context.Context, id string) (*models.Job, error) { + return s.db.GetJob(id) +} + +func (s *JobService) ListJobs(ctx context.Context, filter models.JobFilter) ([]models.Job, error) { + return s.db.ListJobs(filter) +} + +func (s *JobService) CancelJob(ctx context.Context, id string) error { + err := s.db.CancelJob(id) + if err == nil && s.bus != nil { + if j, err := s.db.GetJob(id); err == nil { + s.bus.Publish(ctx, events.Event{ + Type: events.EventJobFailed, + Job: j, + }) + } + } + return err +} + +func (s *JobService) RetryJob(ctx context.Context, id string) error { + j, err := s.db.GetJob(id) + if err != nil { + return err + } + j.Status = "pending" + j.ErrorMessage = "" + j.Attempts = 0 + j.RunAt = time.Now().UTC() + err = s.db.UpdateJob(*j) + if err == nil && s.bus != nil { + s.bus.Publish(ctx, events.Event{ + Type: events.EventJobProgress, + Job: j, + }) + } + return err +} diff --git a/internal/services/settings_service.go b/internal/services/settings_service.go new file mode 100644 index 0000000..e0e178a --- /dev/null +++ b/internal/services/settings_service.go @@ -0,0 +1,98 @@ +package services + +import ( + "context" + "encoding/json" + "fmt" + + "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 SettingsService struct { + db *db.DB + bus *events.EventBus + log *zap.Logger +} + +func NewSettingsService(database *db.DB, bus *events.EventBus) *SettingsService { + return &SettingsService{ + db: database, + bus: bus, + log: logger.Named("settings-service"), + } +} + +func (s *SettingsService) GetPreferences(ctx context.Context, username string) (map[string]interface{}, error) { + user, err := s.db.GetUserByUsername(username) + if err != nil { + return nil, err + } + if user.Preferences == nil { + return make(map[string]interface{}), nil + } + return user.Preferences, nil +} + +func (s *SettingsService) UpdatePreferences(ctx context.Context, username string, newPrefs map[string]interface{}) (map[string]interface{}, error) { + user, err := s.db.GetUserByUsername(username) + if err != nil { + return nil, err + } + if user.Preferences == nil { + user.Preferences = make(map[string]interface{}) + } + for k, v := range newPrefs { + user.Preferences[k] = v + } + if err := s.db.UpdateUserPreferences(user.ID, user.Preferences); err != nil { + return nil, err + } + + if s.bus != nil { + s.bus.Publish(ctx, events.Event{ + Type: events.EventPreferencesSync, + DirectAttrs: user.Preferences, + }) + } + + return user.Preferences, nil +} + +func (s *SettingsService) GetAISettings(ctx context.Context) (*models.AISettings, error) { + raw, err := s.db.GetSetting("ai_settings") + if err != nil || raw == "" { + return &models.AISettings{ + Enabled: false, + Provider: "openai", + BaseURL: "https://api.openai.com/v1", + APIKey: "", + Model: "gpt-4o-mini", + Temperature: 0.3, + MaxTokens: 1000, + Prompts: []models.PromptTemplate{}, + }, nil + } + + var cfg models.AISettings + if err := json.Unmarshal([]byte(raw), &cfg); err != nil { + return nil, fmt.Errorf("failed to parse AI settings: %w", err) + } + return &cfg, nil +} + +func (s *SettingsService) UpdateAISettings(ctx context.Context, cfg models.AISettings) (*models.AISettings, error) { + bytes, err := json.Marshal(cfg) + if err != nil { + return nil, fmt.Errorf("failed to marshal AI settings: %w", err) + } + + if err := s.db.SetSetting("ai_settings", string(bytes)); err != nil { + return nil, err + } + + return &cfg, nil +} diff --git a/internal/worker/engine.go b/internal/worker/engine.go new file mode 100644 index 0000000..0ae48bf --- /dev/null +++ b/internal/worker/engine.go @@ -0,0 +1,268 @@ +package worker + +import ( + "context" + "fmt" + "sync" + "time" + + "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 JobHandlerFunc func(ctx context.Context, job models.Job) (*models.PatchEntryRequest, map[string]interface{}, error) + +type EntryPatcher interface { + PatchEntry(ctx context.Context, entryID string, patch models.PatchEntryRequest) (*models.FeedEntry, *models.EntryDiff, error) + GetEntry(ctx context.Context, id string) (*models.FeedEntry, error) +} + +type JobEngine struct { + db *db.DB + bus *events.EventBus + patcher EntryPatcher + handlers map[string]JobHandlerFunc + semaphores map[string]chan struct{} + maxWorkers int + pollTicker *time.Ticker + wg sync.WaitGroup + log *zap.Logger + mu sync.RWMutex +} + +func NewJobEngine(database *db.DB, bus *events.EventBus, patcher EntryPatcher, maxWorkers int) *JobEngine { + if maxWorkers <= 0 { + maxWorkers = 10 + } + + engine := &JobEngine{ + db: database, + bus: bus, + patcher: patcher, + handlers: make(map[string]JobHandlerFunc), + semaphores: make(map[string]chan struct{}), + maxWorkers: maxWorkers, + log: logger.Named("job-engine"), + } + + // Set default concurrency limits per task type + engine.SetConcurrency("ai_summary", 2) + engine.SetConcurrency("ai_enrich", 2) + engine.SetConcurrency("readability_scrape", 5) + engine.SetConcurrency("tts_generate", 2) + engine.SetConcurrency("feed_refresh", 10) + engine.SetConcurrency("add_tags", 10) + + return engine +} + +func (e *JobEngine) SetConcurrency(jobType string, limit int) { + e.mu.Lock() + defer e.mu.Unlock() + if limit <= 0 { + limit = 2 + } + e.semaphores[jobType] = make(chan struct{}, limit) +} + +func (e *JobEngine) RegisterHandler(jobType string, handler JobHandlerFunc) { + e.mu.Lock() + defer e.mu.Unlock() + e.handlers[jobType] = handler + if _, ok := e.semaphores[jobType]; !ok { + e.semaphores[jobType] = make(chan struct{}, 5) + } + e.log.Info("Registered job handler", zap.String("type", jobType)) +} + +func (e *JobEngine) Start(ctx context.Context) { + e.log.Info("Starting persistent SQLite job queue engine") + + // 1. Recover crashed/orphaned jobs on boot + if err := e.db.ResetRunningJobsToPending(); err != nil { + e.log.Error("Failed to reset orphaned running jobs", zap.Error(err)) + } + + // 2. Start dispatcher loop + e.pollTicker = time.NewTicker(500 * time.Millisecond) + go func() { + for { + select { + case <-ctx.Done(): + return + case <-e.pollTicker.C: + e.dispatchPendingJobs(ctx) + } + } + }() +} + +func (e *JobEngine) Stop() { + if e.pollTicker != nil { + e.pollTicker.Stop() + } + e.wg.Wait() + e.log.Info("Job engine stopped") +} + +func (e *JobEngine) dispatchPendingJobs(ctx context.Context) { + e.mu.RLock() + var registeredTypes []string + for t := range e.handlers { + registeredTypes = append(registeredTypes, t) + } + e.mu.RUnlock() + + if len(registeredTypes) == 0 { + return + } + + // Claim next due job + job, err := e.db.ClaimNextJob(registeredTypes) + if err != nil || job == nil { + return + } + + // Acquire concurrency slot for this job type + e.mu.RLock() + sem, hasSem := e.semaphores[job.Type] + handler, hasHandler := e.handlers[job.Type] + e.mu.RUnlock() + + if !hasHandler { + e.log.Warn("No handler registered for claimed job", zap.String("job_id", job.ID), zap.String("type", job.Type)) + return + } + + if hasSem { + select { + case sem <- struct{}{}: + // Slot acquired + default: + // Concurrency limit reached for this type, release job back to pending + job.Status = "pending" + _ = e.db.UpdateJob(*job) + return + } + } + + e.wg.Add(1) + go func(j models.Job, h JobHandlerFunc, s chan struct{}) { + defer e.wg.Done() + if s != nil { + defer func() { <-s }() + } + + e.executeJob(ctx, j, h) + }(*job, handler, sem) +} + +func (e *JobEngine) executeJob(ctx context.Context, job models.Job, handler JobHandlerFunc) { + e.log.Info("Executing job", + zap.String("job_id", job.ID), + zap.String("type", job.Type), + zap.Int("attempt", job.Attempts+1), + ) + + // Emit Job Progress + if e.bus != nil { + e.bus.Publish(ctx, events.Event{ + Type: events.EventJobProgress, + Job: &job, + }) + } + + start := time.Now() + patchReq, result, err := handler(ctx, job) + duration := time.Since(start) + + if err != nil { + job.Attempts++ + e.log.Error("Job execution failed", + zap.String("job_id", job.ID), + zap.String("type", job.Type), + zap.Int("attempts", job.Attempts), + zap.Duration("duration", duration), + zap.Error(err), + ) + + if job.Attempts < job.MaxAttempts { + // Exponential backoff retry: 5s, 10s, 20s... + backoff := time.Duration(1<