haopengzhan
ebcaa5d72a
- Implement atomic state transitions: add EntryService.PatchEntry() with EntryDiff computation and EventEntryTransition event emission. - Add persistent SQLite task queue (JobEngine): add jobs table with crash recovery, exponential backoff retries, concurrency control, and idempotency key deduplication. - Decompose monolithic HTTP handlers: split handlers.go into domain handlers (auth, feeds, entries, jobs, settings, rules, sse). - Implement real-time Server-Sent Events (/api/events/stream) for multi-tab/device state synchronization. - Fix SQLite concurrency: apply busy_timeout(10000) and journal_mode(WAL) via DSN query parameters to resolve SQLITE_BUSY (5). - Fix AI settings & prompt loading: unmarshal JSON from ai_settings, apply configured prompt templates, and enforce strict enabled/API key validations. - Fix undo share state: remove omitempty from ShareUUID to serialize null for state resets. - Fix middle list counter: compute true unread/total stats from DB and clean switcher button labels. - Adapt mobile UI for iOS: add viewport-fit=cover in index.html and calibrate safe-area insets for the bottom navigation and article reader toolbar. - Add architectural documentation and worklog in docs/009_worklog.md.
127 lines
3.2 KiB
Go
127 lines
3.2 KiB
Go
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
|
|
}
|