7009b24568
- Add bookstore.NotificationStore interface and wire it directly to *storage.Store in Dependencies, bypassing the Asynq wrapper that caused Producer.(*storage.Store) to fail with a 500 on every notification endpoint when Redis is configured - Replace s.deps.Producer.(*storage.Store) type assertion in all 5 notification handlers with s.deps.NotificationStore (nil-safe, always works) - Fix PocketBase filter single-quote bug in ListNotifications, ClearAllNotifications, and MarkAllNotificationsRead (PocketBase requires double quotes for string values)
1695 lines
56 KiB
Go
1695 lines
56 KiB
Go
package storage
|
|
|
|
import (
|
|
"context"
|
|
"encoding/json"
|
|
"fmt"
|
|
"io"
|
|
"log/slog"
|
|
"strings"
|
|
"time"
|
|
|
|
"github.com/libnovel/backend/internal/bookstore"
|
|
"github.com/libnovel/backend/internal/config"
|
|
"github.com/libnovel/backend/internal/domain"
|
|
"github.com/libnovel/backend/internal/taskqueue"
|
|
)
|
|
|
|
// Store is the unified persistence implementation that satisfies all bookstore
|
|
// and taskqueue interfaces. It routes structured data to PocketBase and binary
|
|
// blobs to MinIO.
|
|
type Store struct {
|
|
pb *pbClient
|
|
mc *minioClient
|
|
log *slog.Logger
|
|
}
|
|
|
|
// NewStore initialises PocketBase and MinIO connections and ensures all MinIO
|
|
// buckets exist. Returns a ready-to-use Store.
|
|
func NewStore(ctx context.Context, cfg config.Config, log *slog.Logger) (*Store, error) {
|
|
pb := newPBClient(cfg.PocketBase, log)
|
|
// Validate PocketBase connectivity by fetching an auth token.
|
|
if _, err := pb.authToken(ctx); err != nil {
|
|
return nil, fmt.Errorf("pocketbase: %w", err)
|
|
}
|
|
|
|
mc, err := newMinioClient(cfg.MinIO)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("minio: %w", err)
|
|
}
|
|
if err := mc.ensureBuckets(ctx); err != nil {
|
|
return nil, fmt.Errorf("minio: ensure buckets: %w", err)
|
|
}
|
|
|
|
return &Store{pb: pb, mc: mc, log: log}, nil
|
|
}
|
|
|
|
// Compile-time interface satisfaction.
|
|
var _ bookstore.BookWriter = (*Store)(nil)
|
|
var _ bookstore.BookReader = (*Store)(nil)
|
|
var _ bookstore.RankingStore = (*Store)(nil)
|
|
var _ bookstore.AudioStore = (*Store)(nil)
|
|
var _ bookstore.PresignStore = (*Store)(nil)
|
|
var _ bookstore.ProgressStore = (*Store)(nil)
|
|
var _ bookstore.CoverStore = (*Store)(nil)
|
|
var _ bookstore.TranslationStore = (*Store)(nil)
|
|
var _ bookstore.AIJobStore = (*Store)(nil)
|
|
var _ bookstore.ChapterImageStore = (*Store)(nil)
|
|
var _ bookstore.BookAdminStore = (*Store)(nil)
|
|
var _ taskqueue.Producer = (*Store)(nil)
|
|
var _ taskqueue.Consumer = (*Store)(nil)
|
|
var _ taskqueue.Reader = (*Store)(nil)
|
|
|
|
// ── BookWriter ────────────────────────────────────────────────────────────────
|
|
|
|
func (s *Store) WriteMetadata(ctx context.Context, meta domain.BookMeta) error {
|
|
payload := map[string]any{
|
|
"slug": meta.Slug,
|
|
"title": meta.Title,
|
|
"author": meta.Author,
|
|
"cover": meta.Cover,
|
|
"status": meta.Status,
|
|
"genres": meta.Genres,
|
|
"summary": meta.Summary,
|
|
"total_chapters": meta.TotalChapters,
|
|
"source_url": meta.SourceURL,
|
|
"ranking": meta.Ranking,
|
|
"rating": meta.Rating,
|
|
}
|
|
// Upsert via filter: if exists PATCH, otherwise POST.
|
|
// Use a conflict-retry pattern to handle concurrent scrapes racing to insert
|
|
// the same slug: if POST fails (or another concurrent writer beat us to it),
|
|
// re-fetch and PATCH instead.
|
|
existing, err := s.getBookBySlug(ctx, meta.Slug)
|
|
if err != nil && err != ErrNotFound {
|
|
return fmt.Errorf("WriteMetadata: %w", err)
|
|
}
|
|
if err == ErrNotFound {
|
|
postErr := s.pb.post(ctx, "/api/collections/books/records", payload, nil)
|
|
if postErr == nil {
|
|
return nil
|
|
}
|
|
// POST failed — a concurrent writer may have inserted the same slug.
|
|
// Re-fetch and fall through to PATCH.
|
|
existing, err = s.getBookBySlug(ctx, meta.Slug)
|
|
if err != nil {
|
|
return postErr // original POST error is more informative
|
|
}
|
|
}
|
|
return s.pb.patch(ctx, fmt.Sprintf("/api/collections/books/records/%s", existing.ID), payload)
|
|
}
|
|
|
|
func (s *Store) WriteChapter(ctx context.Context, slug string, chapter domain.Chapter) error {
|
|
key := ChapterObjectKey(slug, chapter.Ref.Number)
|
|
if err := s.mc.putObject(ctx, s.mc.bucketChapters, key, "text/markdown", []byte(chapter.Text)); err != nil {
|
|
return fmt.Errorf("WriteChapter: minio: %w", err)
|
|
}
|
|
// Upsert the chapters_idx record in PocketBase.
|
|
return s.upsertChapterIdx(ctx, slug, chapter.Ref)
|
|
}
|
|
|
|
func (s *Store) WriteChapterRefs(ctx context.Context, slug string, refs []domain.ChapterRef) error {
|
|
for _, ref := range refs {
|
|
if err := s.upsertChapterIdx(ctx, slug, ref); err != nil {
|
|
s.log.Warn("WriteChapterRefs: upsert failed", "slug", slug, "chapter", ref.Number, "err", err)
|
|
}
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func (s *Store) ChapterExists(ctx context.Context, slug string, ref domain.ChapterRef) bool {
|
|
return s.mc.objectExists(ctx, s.mc.bucketChapters, ChapterObjectKey(slug, ref.Number))
|
|
}
|
|
|
|
func (s *Store) upsertChapterIdx(ctx context.Context, slug string, ref domain.ChapterRef) error {
|
|
payload := map[string]any{
|
|
"slug": slug,
|
|
"number": ref.Number,
|
|
"title": ref.Title,
|
|
}
|
|
filter := fmt.Sprintf(`slug=%q&&number=%d`, slug, ref.Number)
|
|
items, err := s.pb.listAll(ctx, "chapters_idx", filter, "")
|
|
if err != nil && err != ErrNotFound {
|
|
return err
|
|
}
|
|
if len(items) == 0 {
|
|
// Set created timestamp on first insert so recentlyUpdatedBooks can sort by it.
|
|
insertPayload := map[string]any{
|
|
"slug": slug,
|
|
"number": ref.Number,
|
|
"title": ref.Title,
|
|
"created": time.Now().UTC().Format(time.RFC3339),
|
|
}
|
|
postErr := s.pb.post(ctx, "/api/collections/chapters_idx/records", insertPayload, nil)
|
|
if postErr == nil {
|
|
return nil
|
|
}
|
|
// POST failed — a concurrent writer may have inserted the same slug+number.
|
|
// Re-fetch and fall through to PATCH (mirrors WriteMetadata retry pattern).
|
|
items, err = s.pb.listAll(ctx, "chapters_idx", filter, "")
|
|
if err != nil || len(items) == 0 {
|
|
return postErr // original POST error is more informative
|
|
}
|
|
}
|
|
var rec struct {
|
|
ID string `json:"id"`
|
|
}
|
|
json.Unmarshal(items[0], &rec)
|
|
return s.pb.patch(ctx, fmt.Sprintf("/api/collections/chapters_idx/records/%s", rec.ID), payload)
|
|
}
|
|
|
|
// DeduplicateChapters removes duplicate chapters_idx records for slug.
|
|
// For each chapter number that has more than one record, it keeps the record
|
|
// with the latest "updated" timestamp and deletes the rest.
|
|
// Returns the number of records deleted.
|
|
func (s *Store) DeduplicateChapters(ctx context.Context, slug string) (int, error) {
|
|
filter := fmt.Sprintf(`slug=%q`, slug)
|
|
items, err := s.pb.listAll(ctx, "chapters_idx", filter, "number")
|
|
if err != nil {
|
|
return 0, fmt.Errorf("DeduplicateChapters: list: %w", err)
|
|
}
|
|
|
|
type record struct {
|
|
ID string `json:"id"`
|
|
Number int `json:"number"`
|
|
Updated string `json:"updated"`
|
|
}
|
|
|
|
// Group records by chapter number.
|
|
byNumber := make(map[int][]record)
|
|
for _, raw := range items {
|
|
var rec record
|
|
if err := json.Unmarshal(raw, &rec); err != nil || rec.ID == "" {
|
|
continue
|
|
}
|
|
byNumber[rec.Number] = append(byNumber[rec.Number], rec)
|
|
}
|
|
|
|
deleted := 0
|
|
for _, recs := range byNumber {
|
|
if len(recs) <= 1 {
|
|
continue
|
|
}
|
|
// Keep the record with the latest Updated timestamp; delete the rest.
|
|
keep := 0
|
|
for i := 1; i < len(recs); i++ {
|
|
if recs[i].Updated > recs[keep].Updated {
|
|
keep = i
|
|
}
|
|
}
|
|
for i, rec := range recs {
|
|
if i == keep {
|
|
continue
|
|
}
|
|
if delErr := s.pb.delete(ctx, fmt.Sprintf("/api/collections/chapters_idx/records/%s", rec.ID)); delErr != nil {
|
|
s.log.Warn("DeduplicateChapters: delete failed", "slug", slug, "number", rec.Number, "id", rec.ID, "err", delErr)
|
|
continue
|
|
}
|
|
deleted++
|
|
}
|
|
}
|
|
return deleted, nil
|
|
}
|
|
|
|
// ── BookReader ────────────────────────────────────────────────────────────────
|
|
|
|
type pbBook struct {
|
|
ID string `json:"id"`
|
|
Slug string `json:"slug"`
|
|
Title string `json:"title"`
|
|
Author string `json:"author"`
|
|
Cover string `json:"cover"`
|
|
Status string `json:"status"`
|
|
Genres []string `json:"genres"`
|
|
Summary string `json:"summary"`
|
|
TotalChapters int `json:"total_chapters"`
|
|
SourceURL string `json:"source_url"`
|
|
Ranking int `json:"ranking"`
|
|
Rating float64 `json:"rating"`
|
|
Updated string `json:"updated"`
|
|
Archived bool `json:"archived"`
|
|
}
|
|
|
|
func (b pbBook) toDomain() domain.BookMeta {
|
|
var metaUpdated int64
|
|
if t, err := time.Parse(time.RFC3339, b.Updated); err == nil {
|
|
metaUpdated = t.Unix()
|
|
}
|
|
return domain.BookMeta{
|
|
Slug: b.Slug,
|
|
Title: b.Title,
|
|
Author: b.Author,
|
|
Cover: b.Cover,
|
|
Status: b.Status,
|
|
Genres: b.Genres,
|
|
Summary: b.Summary,
|
|
TotalChapters: b.TotalChapters,
|
|
SourceURL: b.SourceURL,
|
|
Ranking: b.Ranking,
|
|
Rating: b.Rating,
|
|
MetaUpdated: metaUpdated,
|
|
Archived: b.Archived,
|
|
}
|
|
}
|
|
|
|
func (s *Store) getBookBySlug(ctx context.Context, slug string) (pbBook, error) {
|
|
filter := fmt.Sprintf(`slug=%q`, slug)
|
|
items, err := s.pb.listAll(ctx, "books", filter, "")
|
|
if err != nil {
|
|
return pbBook{}, err
|
|
}
|
|
if len(items) == 0 {
|
|
return pbBook{}, ErrNotFound
|
|
}
|
|
var b pbBook
|
|
json.Unmarshal(items[0], &b)
|
|
return b, nil
|
|
}
|
|
|
|
func (s *Store) ReadMetadata(ctx context.Context, slug string) (domain.BookMeta, bool, error) {
|
|
b, err := s.getBookBySlug(ctx, slug)
|
|
if err == ErrNotFound {
|
|
return domain.BookMeta{}, false, nil
|
|
}
|
|
if err != nil {
|
|
return domain.BookMeta{}, false, err
|
|
}
|
|
return b.toDomain(), true, nil
|
|
}
|
|
|
|
func (s *Store) ListBooks(ctx context.Context) ([]domain.BookMeta, error) {
|
|
items, err := s.pb.listAll(ctx, "books", "archived=false", "title")
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
books := make([]domain.BookMeta, 0, len(items))
|
|
for _, raw := range items {
|
|
var b pbBook
|
|
json.Unmarshal(raw, &b)
|
|
books = append(books, b.toDomain())
|
|
}
|
|
return books, nil
|
|
}
|
|
|
|
func (s *Store) LocalSlugs(ctx context.Context) (map[string]bool, error) {
|
|
items, err := s.pb.listAll(ctx, "books", "", "")
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
slugs := make(map[string]bool, len(items))
|
|
for _, raw := range items {
|
|
var b struct {
|
|
Slug string `json:"slug"`
|
|
}
|
|
json.Unmarshal(raw, &b)
|
|
if b.Slug != "" {
|
|
slugs[b.Slug] = true
|
|
}
|
|
}
|
|
return slugs, nil
|
|
}
|
|
|
|
func (s *Store) MetadataMtime(ctx context.Context, slug string) int64 {
|
|
b, err := s.getBookBySlug(ctx, slug)
|
|
if err != nil {
|
|
return 0
|
|
}
|
|
t, err := time.Parse(time.RFC3339, b.Updated)
|
|
if err != nil {
|
|
return 0
|
|
}
|
|
return t.Unix()
|
|
}
|
|
|
|
func (s *Store) ReadChapter(ctx context.Context, slug string, n int) (string, error) {
|
|
data, err := s.mc.getObject(ctx, s.mc.bucketChapters, ChapterObjectKey(slug, n))
|
|
if err != nil {
|
|
return "", fmt.Errorf("ReadChapter: %w", err)
|
|
}
|
|
return string(data), nil
|
|
}
|
|
|
|
func (s *Store) ListChapters(ctx context.Context, slug string) ([]domain.ChapterInfo, error) {
|
|
filter := fmt.Sprintf(`slug=%q`, slug)
|
|
items, err := s.pb.listAll(ctx, "chapters_idx", filter, "number")
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
chapters := make([]domain.ChapterInfo, 0, len(items))
|
|
for _, raw := range items {
|
|
var rec struct {
|
|
Number int `json:"number"`
|
|
Title string `json:"title"`
|
|
}
|
|
json.Unmarshal(raw, &rec)
|
|
chapters = append(chapters, domain.ChapterInfo{Number: rec.Number, Title: rec.Title})
|
|
}
|
|
return chapters, nil
|
|
}
|
|
|
|
func (s *Store) CountChapters(ctx context.Context, slug string) int {
|
|
chapters, err := s.ListChapters(ctx, slug)
|
|
if err != nil {
|
|
return 0
|
|
}
|
|
return len(chapters)
|
|
}
|
|
|
|
func (s *Store) ReindexChapters(ctx context.Context, slug string) (int, error) {
|
|
keys, err := s.mc.listObjectKeys(ctx, s.mc.bucketChapters, slug+"/")
|
|
if err != nil {
|
|
return 0, fmt.Errorf("ReindexChapters: list objects: %w", err)
|
|
}
|
|
count := 0
|
|
for _, key := range keys {
|
|
if !strings.HasSuffix(key, ".md") {
|
|
continue
|
|
}
|
|
n := chapterNumberFromKey(key)
|
|
if n == 0 {
|
|
continue
|
|
}
|
|
ref := domain.ChapterRef{Number: n}
|
|
if err := s.upsertChapterIdx(ctx, slug, ref); err != nil {
|
|
s.log.Warn("ReindexChapters: upsert failed", "key", key, "err", err)
|
|
continue
|
|
}
|
|
count++
|
|
}
|
|
return count, nil
|
|
}
|
|
|
|
// ── BookAdminStore ────────────────────────────────────────────────────────────
|
|
|
|
// ArchiveBook sets archived=true on the book record for slug.
|
|
func (s *Store) ArchiveBook(ctx context.Context, slug string) error {
|
|
book, err := s.getBookBySlug(ctx, slug)
|
|
if err == ErrNotFound {
|
|
return ErrNotFound
|
|
}
|
|
if err != nil {
|
|
return fmt.Errorf("ArchiveBook: %w", err)
|
|
}
|
|
return s.pb.patch(ctx, fmt.Sprintf("/api/collections/books/records/%s", book.ID),
|
|
map[string]any{"archived": true})
|
|
}
|
|
|
|
// UnarchiveBook clears archived on the book record for slug.
|
|
func (s *Store) UnarchiveBook(ctx context.Context, slug string) error {
|
|
book, err := s.getBookBySlug(ctx, slug)
|
|
if err == ErrNotFound {
|
|
return ErrNotFound
|
|
}
|
|
if err != nil {
|
|
return fmt.Errorf("UnarchiveBook: %w", err)
|
|
}
|
|
return s.pb.patch(ctx, fmt.Sprintf("/api/collections/books/records/%s", book.ID),
|
|
map[string]any{"archived": false})
|
|
}
|
|
|
|
// DeleteBook permanently removes all data for a book:
|
|
// - PocketBase books record
|
|
// - All PocketBase chapters_idx records for the slug
|
|
// - All MinIO chapter markdown objects ({slug}/chapter-*.md)
|
|
// - MinIO cover image (covers/{slug}.jpg)
|
|
func (s *Store) DeleteBook(ctx context.Context, slug string) error {
|
|
// 1. Fetch the book record to get its PocketBase ID.
|
|
book, err := s.getBookBySlug(ctx, slug)
|
|
if err == ErrNotFound {
|
|
return ErrNotFound
|
|
}
|
|
if err != nil {
|
|
return fmt.Errorf("DeleteBook: fetch: %w", err)
|
|
}
|
|
|
|
// 2. Delete all chapters_idx records.
|
|
filter := fmt.Sprintf(`slug=%q`, slug)
|
|
items, err := s.pb.listAll(ctx, "chapters_idx", filter, "")
|
|
if err != nil && err != ErrNotFound {
|
|
return fmt.Errorf("DeleteBook: list chapters_idx: %w", err)
|
|
}
|
|
for _, raw := range items {
|
|
var rec struct {
|
|
ID string `json:"id"`
|
|
}
|
|
if json.Unmarshal(raw, &rec) == nil && rec.ID != "" {
|
|
if delErr := s.pb.delete(ctx, fmt.Sprintf("/api/collections/chapters_idx/records/%s", rec.ID)); delErr != nil {
|
|
s.log.Warn("DeleteBook: delete chapters_idx record failed", "slug", slug, "id", rec.ID, "err", delErr)
|
|
}
|
|
}
|
|
}
|
|
|
|
// 3. Delete MinIO chapter objects.
|
|
if err := s.mc.deleteObjects(ctx, s.mc.bucketChapters, slug+"/"); err != nil {
|
|
s.log.Warn("DeleteBook: delete chapter objects failed", "slug", slug, "err", err)
|
|
}
|
|
|
|
// 4. Delete MinIO cover image.
|
|
if err := s.mc.deleteObjects(ctx, s.mc.bucketBrowse, CoverObjectKey(slug)); err != nil {
|
|
s.log.Warn("DeleteBook: delete cover failed", "slug", slug, "err", err)
|
|
}
|
|
|
|
// 5. Delete the PocketBase books record.
|
|
if err := s.pb.delete(ctx, fmt.Sprintf("/api/collections/books/records/%s", book.ID)); err != nil {
|
|
return fmt.Errorf("DeleteBook: delete books record: %w", err)
|
|
}
|
|
|
|
return nil
|
|
}
|
|
|
|
// ── RankingStore ──────────────────────────────────────────────────────────────
|
|
|
|
func (s *Store) WriteRankingItem(ctx context.Context, item domain.RankingItem) error {
|
|
payload := map[string]any{
|
|
"rank": item.Rank,
|
|
"slug": item.Slug,
|
|
"title": item.Title,
|
|
"author": item.Author,
|
|
"cover": item.Cover,
|
|
"status": item.Status,
|
|
"genres": item.Genres,
|
|
"source_url": item.SourceURL,
|
|
}
|
|
filter := fmt.Sprintf(`slug=%q`, item.Slug)
|
|
items, err := s.pb.listAll(ctx, "ranking", filter, "")
|
|
if err != nil && err != ErrNotFound {
|
|
return err
|
|
}
|
|
if len(items) == 0 {
|
|
return s.pb.post(ctx, "/api/collections/ranking/records", payload, nil)
|
|
}
|
|
var rec struct {
|
|
ID string `json:"id"`
|
|
}
|
|
json.Unmarshal(items[0], &rec)
|
|
return s.pb.patch(ctx, fmt.Sprintf("/api/collections/ranking/records/%s", rec.ID), payload)
|
|
}
|
|
|
|
func (s *Store) ReadRankingItems(ctx context.Context) ([]domain.RankingItem, error) {
|
|
items, err := s.pb.listAll(ctx, "ranking", "", "rank")
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
result := make([]domain.RankingItem, 0, len(items))
|
|
for _, raw := range items {
|
|
var rec struct {
|
|
Rank int `json:"rank"`
|
|
Slug string `json:"slug"`
|
|
Title string `json:"title"`
|
|
Author string `json:"author"`
|
|
Cover string `json:"cover"`
|
|
Status string `json:"status"`
|
|
Genres []string `json:"genres"`
|
|
SourceURL string `json:"source_url"`
|
|
Updated string `json:"updated"`
|
|
}
|
|
json.Unmarshal(raw, &rec)
|
|
t, _ := time.Parse(time.RFC3339, rec.Updated)
|
|
result = append(result, domain.RankingItem{
|
|
Rank: rec.Rank,
|
|
Slug: rec.Slug,
|
|
Title: rec.Title,
|
|
Author: rec.Author,
|
|
Cover: rec.Cover,
|
|
Status: rec.Status,
|
|
Genres: rec.Genres,
|
|
SourceURL: rec.SourceURL,
|
|
Updated: t,
|
|
})
|
|
}
|
|
return result, nil
|
|
}
|
|
|
|
func (s *Store) RankingFreshEnough(ctx context.Context, maxAge time.Duration) (bool, error) {
|
|
items, err := s.ReadRankingItems(ctx)
|
|
if err != nil || len(items) == 0 {
|
|
return false, err
|
|
}
|
|
var latest time.Time
|
|
for _, item := range items {
|
|
if item.Updated.After(latest) {
|
|
latest = item.Updated
|
|
}
|
|
}
|
|
return time.Since(latest) < maxAge, nil
|
|
}
|
|
|
|
// ── AudioStore ────────────────────────────────────────────────────────────────
|
|
|
|
func (s *Store) AudioObjectKey(slug string, n int, voice string) string {
|
|
return AudioObjectKey(slug, n, voice)
|
|
}
|
|
|
|
func (s *Store) AudioObjectKeyExt(slug string, n int, voice, ext string) string {
|
|
return AudioObjectKeyExt(slug, n, voice, ext)
|
|
}
|
|
|
|
func (s *Store) AudioExists(ctx context.Context, key string) bool {
|
|
return s.mc.objectExists(ctx, s.mc.bucketAudio, key)
|
|
}
|
|
|
|
func (s *Store) PutAudio(ctx context.Context, key string, data []byte) error {
|
|
return s.mc.putObject(ctx, s.mc.bucketAudio, key, "audio/mpeg", data)
|
|
}
|
|
|
|
func (s *Store) PutAudioStream(ctx context.Context, key string, r io.Reader, size int64, contentType string) error {
|
|
return s.mc.putObjectStream(ctx, s.mc.bucketAudio, key, contentType, r, size)
|
|
}
|
|
|
|
// ── PresignStore ──────────────────────────────────────────────────────────────
|
|
|
|
func (s *Store) PresignChapter(ctx context.Context, slug string, n int, expires time.Duration) (string, error) {
|
|
return s.mc.presignGet(ctx, s.mc.bucketChapters, ChapterObjectKey(slug, n), expires)
|
|
}
|
|
|
|
func (s *Store) PresignAudio(ctx context.Context, key string, expires time.Duration) (string, error) {
|
|
return s.mc.presignGet(ctx, s.mc.bucketAudio, key, expires)
|
|
}
|
|
|
|
func (s *Store) PresignAvatarUpload(ctx context.Context, userID, ext string) (uploadURL, key string, err error) {
|
|
key = AvatarObjectKey(userID, ext)
|
|
uploadURL, err = s.mc.presignPut(ctx, s.mc.bucketAvatars, key, 15*time.Minute)
|
|
return
|
|
}
|
|
|
|
func (s *Store) PresignAvatarURL(ctx context.Context, userID string) (string, bool, error) {
|
|
for _, ext := range []string{"jpg", "png", "webp"} {
|
|
key := AvatarObjectKey(userID, ext)
|
|
if s.mc.objectExists(ctx, s.mc.bucketAvatars, key) {
|
|
u, err := s.mc.presignGet(ctx, s.mc.bucketAvatars, key, 1*time.Hour)
|
|
return u, true, err
|
|
}
|
|
}
|
|
return "", false, nil
|
|
}
|
|
|
|
func (s *Store) PutAvatar(ctx context.Context, userID, ext, contentType string, data []byte) (string, error) {
|
|
// Delete existing avatar objects for this user before writing the new one
|
|
// so old extensions don't linger (e.g. old .png after uploading a .jpg).
|
|
_ = s.mc.deleteObjects(ctx, s.mc.bucketAvatars, userID+"/")
|
|
key := AvatarObjectKey(userID, ext)
|
|
if err := s.mc.putObject(ctx, s.mc.bucketAvatars, key, contentType, data); err != nil {
|
|
return "", fmt.Errorf("put avatar: %w", err)
|
|
}
|
|
return key, nil
|
|
}
|
|
|
|
func (s *Store) DeleteAvatar(ctx context.Context, userID string) error {
|
|
return s.mc.deleteObjects(ctx, s.mc.bucketAvatars, userID+"/")
|
|
}
|
|
|
|
// ── ProgressStore ─────────────────────────────────────────────────────────────
|
|
|
|
func (s *Store) GetProgress(ctx context.Context, sessionID, slug string) (domain.ReadingProgress, bool) {
|
|
filter := fmt.Sprintf(`session_id=%q&&slug=%q`, sessionID, slug)
|
|
items, err := s.pb.listAll(ctx, "progress", filter, "")
|
|
if err != nil || len(items) == 0 {
|
|
return domain.ReadingProgress{}, false
|
|
}
|
|
var rec struct {
|
|
Slug string `json:"slug"`
|
|
Chapter int `json:"chapter"`
|
|
UpdatedAt string `json:"updated"`
|
|
}
|
|
json.Unmarshal(items[0], &rec)
|
|
t, _ := time.Parse(time.RFC3339, rec.UpdatedAt)
|
|
return domain.ReadingProgress{Slug: rec.Slug, Chapter: rec.Chapter, UpdatedAt: t}, true
|
|
}
|
|
|
|
func (s *Store) SetProgress(ctx context.Context, sessionID string, p domain.ReadingProgress) error {
|
|
payload := map[string]any{
|
|
"session_id": sessionID,
|
|
"slug": p.Slug,
|
|
"chapter": p.Chapter,
|
|
}
|
|
filter := fmt.Sprintf(`session_id=%q&&slug=%q`, sessionID, p.Slug)
|
|
items, err := s.pb.listAll(ctx, "progress", filter, "")
|
|
if err != nil && err != ErrNotFound {
|
|
return err
|
|
}
|
|
if len(items) == 0 {
|
|
return s.pb.post(ctx, "/api/collections/progress/records", payload, nil)
|
|
}
|
|
var rec struct {
|
|
ID string `json:"id"`
|
|
}
|
|
json.Unmarshal(items[0], &rec)
|
|
return s.pb.patch(ctx, fmt.Sprintf("/api/collections/progress/records/%s", rec.ID), payload)
|
|
}
|
|
|
|
func (s *Store) AllProgress(ctx context.Context, sessionID string) ([]domain.ReadingProgress, error) {
|
|
filter := fmt.Sprintf(`session_id=%q`, sessionID)
|
|
items, err := s.pb.listAll(ctx, "progress", filter, "-updated")
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
result := make([]domain.ReadingProgress, 0, len(items))
|
|
for _, raw := range items {
|
|
var rec struct {
|
|
Slug string `json:"slug"`
|
|
Chapter int `json:"chapter"`
|
|
UpdatedAt string `json:"updated"`
|
|
}
|
|
json.Unmarshal(raw, &rec)
|
|
t, _ := time.Parse(time.RFC3339, rec.UpdatedAt)
|
|
result = append(result, domain.ReadingProgress{Slug: rec.Slug, Chapter: rec.Chapter, UpdatedAt: t})
|
|
}
|
|
return result, nil
|
|
}
|
|
|
|
func (s *Store) DeleteProgress(ctx context.Context, sessionID, slug string) error {
|
|
filter := fmt.Sprintf(`session_id=%q&&slug=%q`, sessionID, slug)
|
|
items, err := s.pb.listAll(ctx, "progress", filter, "")
|
|
if err != nil || len(items) == 0 {
|
|
return nil
|
|
}
|
|
var rec struct {
|
|
ID string `json:"id"`
|
|
}
|
|
json.Unmarshal(items[0], &rec)
|
|
return s.pb.delete(ctx, fmt.Sprintf("/api/collections/progress/records/%s", rec.ID))
|
|
}
|
|
|
|
// ── taskqueue.Producer ────────────────────────────────────────────────────────
|
|
|
|
func (s *Store) CreateScrapeTask(ctx context.Context, kind, targetURL string, fromChapter, toChapter int) (string, error) {
|
|
payload := map[string]any{
|
|
"kind": kind,
|
|
"target_url": targetURL,
|
|
"from_chapter": fromChapter,
|
|
"to_chapter": toChapter,
|
|
"status": string(domain.TaskStatusPending),
|
|
"started": time.Now().UTC().Format(time.RFC3339),
|
|
}
|
|
var rec struct {
|
|
ID string `json:"id"`
|
|
}
|
|
if err := s.pb.post(ctx, "/api/collections/scraping_tasks/records", payload, &rec); err != nil {
|
|
return "", err
|
|
}
|
|
return rec.ID, nil
|
|
}
|
|
|
|
func (s *Store) CreateAudioTask(ctx context.Context, slug string, chapter int, voice string) (string, error) {
|
|
cacheKey := fmt.Sprintf("%s/%d/%s", slug, chapter, voice)
|
|
payload := map[string]any{
|
|
"cache_key": cacheKey,
|
|
"slug": slug,
|
|
"chapter": chapter,
|
|
"voice": voice,
|
|
"status": string(domain.TaskStatusPending),
|
|
"started": time.Now().UTC().Format(time.RFC3339),
|
|
}
|
|
var rec struct {
|
|
ID string `json:"id"`
|
|
}
|
|
if err := s.pb.post(ctx, "/api/collections/audio_jobs/records", payload, &rec); err != nil {
|
|
return "", err
|
|
}
|
|
return rec.ID, nil
|
|
}
|
|
|
|
func (s *Store) CreateTranslationTask(ctx context.Context, slug string, chapter int, lang string) (string, error) {
|
|
cacheKey := fmt.Sprintf("%s/%d/%s", slug, chapter, lang)
|
|
payload := map[string]any{
|
|
"cache_key": cacheKey,
|
|
"slug": slug,
|
|
"chapter": chapter,
|
|
"lang": lang,
|
|
"status": string(domain.TaskStatusPending),
|
|
"started": time.Now().UTC().Format(time.RFC3339),
|
|
}
|
|
var rec struct {
|
|
ID string `json:"id"`
|
|
}
|
|
if err := s.pb.post(ctx, "/api/collections/translation_jobs/records", payload, &rec); err != nil {
|
|
return "", err
|
|
}
|
|
return rec.ID, nil
|
|
}
|
|
|
|
func (s *Store) CreateImportTask(ctx context.Context, task domain.ImportTask) (string, error) {
|
|
payload := map[string]any{
|
|
"slug": task.Slug,
|
|
"title": task.Title,
|
|
"file_name": task.Slug + "." + task.FileType,
|
|
"file_type": task.FileType,
|
|
"object_key": task.ObjectKey,
|
|
"chapters_key": task.ChaptersKey,
|
|
"author": task.Author,
|
|
"cover_url": task.CoverURL,
|
|
"summary": task.Summary,
|
|
"book_status": task.BookStatus,
|
|
"status": string(domain.TaskStatusPending),
|
|
"chapters_done": 0,
|
|
"chapters_total": task.ChaptersTotal,
|
|
"started": time.Now().UTC().Format(time.RFC3339),
|
|
"initiator_user_id": task.InitiatorUserID,
|
|
}
|
|
if len(task.Genres) > 0 {
|
|
payload["genres"] = strings.Join(task.Genres, ",")
|
|
}
|
|
var rec struct {
|
|
ID string `json:"id"`
|
|
}
|
|
if err := s.pb.post(ctx, "/api/collections/import_tasks/records", payload, &rec); err != nil {
|
|
return "", err
|
|
}
|
|
return rec.ID, nil
|
|
}
|
|
|
|
// CreateNotification creates a notification record in PocketBase.
|
|
func (s *Store) CreateNotification(ctx context.Context, userID, title, message, link string) error {
|
|
payload := map[string]any{
|
|
"user_id": userID,
|
|
"title": title,
|
|
"message": message,
|
|
"link": link,
|
|
"read": false,
|
|
"created": time.Now().UTC().Format(time.RFC3339),
|
|
}
|
|
return s.pb.post(ctx, "/api/collections/notifications/records", payload, nil)
|
|
}
|
|
|
|
// ListNotifications returns notifications for a user.
|
|
func (s *Store) ListNotifications(ctx context.Context, userID string, limit int) ([]map[string]any, error) {
|
|
filter := fmt.Sprintf(`user_id="%s"`, userID)
|
|
items, err := s.pb.listAll(ctx, "notifications", filter, "-created")
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
// Parse each json.RawMessage into a map
|
|
results := make([]map[string]any, 0, len(items))
|
|
for _, raw := range items {
|
|
var m map[string]any
|
|
if json.Unmarshal(raw, &m) == nil {
|
|
results = append(results, m)
|
|
}
|
|
}
|
|
if limit > 0 && len(results) > limit {
|
|
results = results[:limit]
|
|
}
|
|
return results, nil
|
|
}
|
|
|
|
// MarkNotificationRead marks a notification as read.
|
|
func (s *Store) MarkNotificationRead(ctx context.Context, id string) error {
|
|
return s.pb.patch(ctx, fmt.Sprintf("/api/collections/notifications/records/%s", id),
|
|
map[string]any{"read": true})
|
|
}
|
|
|
|
// DeleteNotification deletes a single notification by ID.
|
|
func (s *Store) DeleteNotification(ctx context.Context, id string) error {
|
|
return s.pb.delete(ctx, fmt.Sprintf("/api/collections/notifications/records/%s", id))
|
|
}
|
|
|
|
// ClearAllNotifications deletes all notifications for a user.
|
|
func (s *Store) ClearAllNotifications(ctx context.Context, userID string) error {
|
|
filter := fmt.Sprintf(`user_id="%s"`, userID)
|
|
items, err := s.pb.listAll(ctx, "notifications", filter, "")
|
|
if err != nil {
|
|
return fmt.Errorf("ClearAllNotifications list: %w", err)
|
|
}
|
|
for _, raw := range items {
|
|
var rec struct {
|
|
ID string `json:"id"`
|
|
}
|
|
if json.Unmarshal(raw, &rec) == nil && rec.ID != "" {
|
|
_ = s.pb.delete(ctx, fmt.Sprintf("/api/collections/notifications/records/%s", rec.ID))
|
|
}
|
|
}
|
|
return nil
|
|
}
|
|
|
|
// MarkAllNotificationsRead marks all notifications for a user as read.
|
|
func (s *Store) MarkAllNotificationsRead(ctx context.Context, userID string) error {
|
|
filter := fmt.Sprintf(`user_id="%s"&&read=false`, userID)
|
|
items, err := s.pb.listAll(ctx, "notifications", filter, "")
|
|
if err != nil {
|
|
return fmt.Errorf("MarkAllNotificationsRead list: %w", err)
|
|
}
|
|
for _, raw := range items {
|
|
var rec struct {
|
|
ID string `json:"id"`
|
|
}
|
|
if json.Unmarshal(raw, &rec) == nil && rec.ID != "" {
|
|
_ = s.pb.patch(ctx, fmt.Sprintf("/api/collections/notifications/records/%s", rec.ID),
|
|
map[string]any{"read": true})
|
|
}
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func (s *Store) CancelTask(ctx context.Context, id string) error {
|
|
// Try scraping_tasks first, then audio_jobs, then translation_jobs.
|
|
if err := s.pb.patch(ctx, fmt.Sprintf("/api/collections/scraping_tasks/records/%s", id),
|
|
map[string]string{"status": string(domain.TaskStatusCancelled)}); err == nil {
|
|
return nil
|
|
}
|
|
if err := s.pb.patch(ctx, fmt.Sprintf("/api/collections/audio_jobs/records/%s", id),
|
|
map[string]string{"status": string(domain.TaskStatusCancelled)}); err == nil {
|
|
return nil
|
|
}
|
|
return s.pb.patch(ctx, fmt.Sprintf("/api/collections/translation_jobs/records/%s", id),
|
|
map[string]string{"status": string(domain.TaskStatusCancelled)})
|
|
}
|
|
|
|
func (s *Store) CancelAudioTasksBySlug(ctx context.Context, slug string) (int, error) {
|
|
filter := fmt.Sprintf(`slug='%s'&&(status='pending'||status='running')`, slug)
|
|
items, err := s.pb.listAll(ctx, "audio_jobs", filter, "")
|
|
if err != nil {
|
|
return 0, fmt.Errorf("CancelAudioTasksBySlug list: %w", err)
|
|
}
|
|
cancelled := 0
|
|
for _, raw := range items {
|
|
var rec struct {
|
|
ID string `json:"id"`
|
|
}
|
|
if json.Unmarshal(raw, &rec) == nil && rec.ID != "" {
|
|
if patchErr := s.pb.patch(ctx,
|
|
fmt.Sprintf("/api/collections/audio_jobs/records/%s", rec.ID),
|
|
map[string]string{"status": string(domain.TaskStatusCancelled)}); patchErr == nil {
|
|
cancelled++
|
|
}
|
|
}
|
|
}
|
|
return cancelled, nil
|
|
}
|
|
|
|
// ── taskqueue.Consumer ────────────────────────────────────────────────────────
|
|
|
|
func (s *Store) ClaimNextScrapeTask(ctx context.Context, workerID string) (domain.ScrapeTask, bool, error) {
|
|
raw, err := s.pb.claimRecord(ctx, "scraping_tasks", workerID, nil)
|
|
if err != nil {
|
|
return domain.ScrapeTask{}, false, err
|
|
}
|
|
if raw == nil {
|
|
return domain.ScrapeTask{}, false, nil
|
|
}
|
|
task, err := parseScrapeTask(raw)
|
|
return task, err == nil, err
|
|
}
|
|
|
|
func (s *Store) ClaimNextAudioTask(ctx context.Context, workerID string) (domain.AudioTask, bool, error) {
|
|
raw, err := s.pb.claimRecord(ctx, "audio_jobs", workerID, nil)
|
|
if err != nil {
|
|
return domain.AudioTask{}, false, err
|
|
}
|
|
if raw == nil {
|
|
return domain.AudioTask{}, false, nil
|
|
}
|
|
task, err := parseAudioTask(raw)
|
|
return task, err == nil, err
|
|
}
|
|
|
|
func (s *Store) ClaimNextTranslationTask(ctx context.Context, workerID string) (domain.TranslationTask, bool, error) {
|
|
raw, err := s.pb.claimRecord(ctx, "translation_jobs", workerID, nil)
|
|
if err != nil {
|
|
return domain.TranslationTask{}, false, err
|
|
}
|
|
if raw == nil {
|
|
return domain.TranslationTask{}, false, nil
|
|
}
|
|
task, err := parseTranslationTask(raw)
|
|
return task, err == nil, err
|
|
}
|
|
|
|
func (s *Store) ClaimNextImportTask(ctx context.Context, workerID string) (domain.ImportTask, bool, error) {
|
|
raw, err := s.pb.claimRecord(ctx, "import_tasks", workerID, nil)
|
|
if err != nil {
|
|
return domain.ImportTask{}, false, err
|
|
}
|
|
if raw == nil {
|
|
return domain.ImportTask{}, false, nil
|
|
}
|
|
task, err := parseImportTask(raw)
|
|
return task, err == nil, err
|
|
}
|
|
|
|
func (s *Store) FinishScrapeTask(ctx context.Context, id string, result domain.ScrapeResult) error {
|
|
status := string(domain.TaskStatusDone)
|
|
if result.ErrorMessage != "" {
|
|
status = string(domain.TaskStatusFailed)
|
|
}
|
|
return s.pb.patch(ctx, fmt.Sprintf("/api/collections/scraping_tasks/records/%s", id), map[string]any{
|
|
"status": status,
|
|
"books_found": result.BooksFound,
|
|
"chapters_scraped": result.ChaptersScraped,
|
|
"chapters_skipped": result.ChaptersSkipped,
|
|
"errors": result.Errors,
|
|
"error_message": result.ErrorMessage,
|
|
"finished": time.Now().UTC().Format(time.RFC3339),
|
|
})
|
|
}
|
|
|
|
func (s *Store) FinishAudioTask(ctx context.Context, id string, result domain.AudioResult) error {
|
|
status := string(domain.TaskStatusDone)
|
|
if result.ErrorMessage != "" {
|
|
status = string(domain.TaskStatusFailed)
|
|
}
|
|
return s.pb.patch(ctx, fmt.Sprintf("/api/collections/audio_jobs/records/%s", id), map[string]any{
|
|
"status": status,
|
|
"error_message": result.ErrorMessage,
|
|
"finished": time.Now().UTC().Format(time.RFC3339),
|
|
})
|
|
}
|
|
|
|
func (s *Store) FinishTranslationTask(ctx context.Context, id string, result domain.TranslationResult) error {
|
|
status := string(domain.TaskStatusDone)
|
|
if result.ErrorMessage != "" {
|
|
status = string(domain.TaskStatusFailed)
|
|
}
|
|
return s.pb.patch(ctx, fmt.Sprintf("/api/collections/translation_jobs/records/%s", id), map[string]any{
|
|
"status": status,
|
|
"error_message": result.ErrorMessage,
|
|
"finished": time.Now().UTC().Format(time.RFC3339),
|
|
})
|
|
}
|
|
|
|
func (s *Store) FinishImportTask(ctx context.Context, id string, result domain.ImportResult) error {
|
|
status := string(domain.TaskStatusDone)
|
|
if result.ErrorMessage != "" {
|
|
status = string(domain.TaskStatusFailed)
|
|
}
|
|
return s.pb.patch(ctx, fmt.Sprintf("/api/collections/import_tasks/records/%s", id), map[string]any{
|
|
"status": status,
|
|
"chapters_done": result.ChaptersImported,
|
|
"chapters_total": result.ChaptersImported,
|
|
"error_message": result.ErrorMessage,
|
|
"finished": time.Now().UTC().Format(time.RFC3339),
|
|
})
|
|
}
|
|
|
|
func (s *Store) FailTask(ctx context.Context, id, errMsg string) error {
|
|
payload := map[string]any{
|
|
"status": string(domain.TaskStatusFailed),
|
|
"error_message": errMsg,
|
|
"finished": time.Now().UTC().Format(time.RFC3339),
|
|
}
|
|
if err := s.pb.patch(ctx, fmt.Sprintf("/api/collections/scraping_tasks/records/%s", id), payload); err == nil {
|
|
return nil
|
|
}
|
|
if err := s.pb.patch(ctx, fmt.Sprintf("/api/collections/audio_jobs/records/%s", id), payload); err == nil {
|
|
return nil
|
|
}
|
|
return s.pb.patch(ctx, fmt.Sprintf("/api/collections/translation_jobs/records/%s", id), payload)
|
|
}
|
|
|
|
// HeartbeatTask updates the heartbeat_at field on a running task.
|
|
// Tries scraping_tasks, audio_jobs, translation_jobs, then import_tasks.
|
|
func (s *Store) HeartbeatTask(ctx context.Context, id string) error {
|
|
payload := map[string]any{
|
|
"heartbeat_at": time.Now().UTC().Format(time.RFC3339),
|
|
}
|
|
if err := s.pb.patch(ctx, fmt.Sprintf("/api/collections/scraping_tasks/records/%s", id), payload); err == nil {
|
|
return nil
|
|
}
|
|
if err := s.pb.patch(ctx, fmt.Sprintf("/api/collections/audio_jobs/records/%s", id), payload); err == nil {
|
|
return nil
|
|
}
|
|
if err := s.pb.patch(ctx, fmt.Sprintf("/api/collections/translation_jobs/records/%s", id), payload); err == nil {
|
|
return nil
|
|
}
|
|
return s.pb.patch(ctx, fmt.Sprintf("/api/collections/import_tasks/records/%s", id), payload)
|
|
}
|
|
|
|
// ReapStaleTasks finds all running tasks whose heartbeat_at is either missing
|
|
// or older than staleAfter, and resets them to pending so they can be
|
|
// re-claimed. Returns the number of tasks reaped.
|
|
func (s *Store) ReapStaleTasks(ctx context.Context, staleAfter time.Duration) (int, error) {
|
|
threshold := time.Now().UTC().Add(-staleAfter).Format(time.RFC3339)
|
|
// Match tasks that are running AND (heartbeat_at is null OR heartbeat_at < threshold).
|
|
// PocketBase datetime fields require `=null` not `=""` in filter expressions.
|
|
filter := fmt.Sprintf(`status="running"&&(heartbeat_at=null||heartbeat_at<"%s")`, threshold)
|
|
resetPayload := map[string]any{
|
|
"status": string(domain.TaskStatusPending),
|
|
"worker_id": "",
|
|
"heartbeat_at": nil,
|
|
}
|
|
|
|
total := 0
|
|
for _, collection := range []string{"scraping_tasks", "audio_jobs", "translation_jobs", "import_tasks"} {
|
|
items, err := s.pb.listAll(ctx, collection, filter, "")
|
|
if err != nil {
|
|
return total, fmt.Errorf("ReapStaleTasks list %s: %w", collection, err)
|
|
}
|
|
for _, raw := range items {
|
|
var rec struct {
|
|
ID string `json:"id"`
|
|
}
|
|
if err := json.Unmarshal(raw, &rec); err != nil || rec.ID == "" {
|
|
continue
|
|
}
|
|
path := fmt.Sprintf("/api/collections/%s/records/%s", collection, rec.ID)
|
|
if err := s.pb.patch(ctx, path, resetPayload); err != nil {
|
|
s.log.Warn("ReapStaleTasks: patch failed", "collection", collection, "id", rec.ID, "err", err)
|
|
continue
|
|
}
|
|
total++
|
|
}
|
|
}
|
|
return total, nil
|
|
}
|
|
|
|
// ── taskqueue.Reader ──────────────────────────────────────────────────────────
|
|
|
|
func (s *Store) ListScrapeTasks(ctx context.Context) ([]domain.ScrapeTask, error) {
|
|
items, err := s.pb.listAll(ctx, "scraping_tasks", "", "-started")
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
tasks := make([]domain.ScrapeTask, 0, len(items))
|
|
for _, raw := range items {
|
|
t, err := parseScrapeTask(raw)
|
|
if err == nil {
|
|
tasks = append(tasks, t)
|
|
}
|
|
}
|
|
return tasks, nil
|
|
}
|
|
|
|
func (s *Store) GetScrapeTask(ctx context.Context, id string) (domain.ScrapeTask, bool, error) {
|
|
var raw json.RawMessage
|
|
if err := s.pb.get(ctx, fmt.Sprintf("/api/collections/scraping_tasks/records/%s", id), &raw); err != nil {
|
|
if err == ErrNotFound {
|
|
return domain.ScrapeTask{}, false, nil
|
|
}
|
|
return domain.ScrapeTask{}, false, err
|
|
}
|
|
t, err := parseScrapeTask(raw)
|
|
return t, err == nil, err
|
|
}
|
|
|
|
func (s *Store) ListAudioTasks(ctx context.Context) ([]domain.AudioTask, error) {
|
|
items, err := s.pb.listAll(ctx, "audio_jobs", "", "-started")
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
tasks := make([]domain.AudioTask, 0, len(items))
|
|
for _, raw := range items {
|
|
t, err := parseAudioTask(raw)
|
|
if err == nil {
|
|
tasks = append(tasks, t)
|
|
}
|
|
}
|
|
return tasks, nil
|
|
}
|
|
|
|
func (s *Store) GetAudioTask(ctx context.Context, cacheKey string) (domain.AudioTask, bool, error) {
|
|
filter := fmt.Sprintf(`cache_key='%s'`, cacheKey)
|
|
items, err := s.pb.listAll(ctx, "audio_jobs", filter, "-started")
|
|
if err != nil || len(items) == 0 {
|
|
return domain.AudioTask{}, false, err
|
|
}
|
|
t, err := parseAudioTask(items[0])
|
|
return t, err == nil, err
|
|
}
|
|
|
|
func (s *Store) ListTranslationTasks(ctx context.Context) ([]domain.TranslationTask, error) {
|
|
items, err := s.pb.listAll(ctx, "translation_jobs", "", "-started")
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
tasks := make([]domain.TranslationTask, 0, len(items))
|
|
for _, raw := range items {
|
|
t, err := parseTranslationTask(raw)
|
|
if err == nil {
|
|
tasks = append(tasks, t)
|
|
}
|
|
}
|
|
return tasks, nil
|
|
}
|
|
|
|
func (s *Store) GetTranslationTask(ctx context.Context, cacheKey string) (domain.TranslationTask, bool, error) {
|
|
items, err := s.pb.listAll(ctx, "translation_jobs", fmt.Sprintf("cache_key=%q", cacheKey), "-started")
|
|
if err != nil || len(items) == 0 {
|
|
return domain.TranslationTask{}, false, err
|
|
}
|
|
t, err := parseTranslationTask(items[0])
|
|
return t, err == nil, err
|
|
}
|
|
|
|
func (s *Store) ListImportTasks(ctx context.Context) ([]domain.ImportTask, error) {
|
|
items, err := s.pb.listAll(ctx, "import_tasks", "", "-started")
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
tasks := make([]domain.ImportTask, 0, len(items))
|
|
for _, raw := range items {
|
|
t, err := parseImportTask(raw)
|
|
if err == nil {
|
|
tasks = append(tasks, t)
|
|
}
|
|
}
|
|
return tasks, nil
|
|
}
|
|
|
|
func (s *Store) GetImportTask(ctx context.Context, id string) (domain.ImportTask, bool, error) {
|
|
var raw json.RawMessage
|
|
if err := s.pb.get(ctx, fmt.Sprintf("/api/collections/import_tasks/records/%s", id), &raw); err != nil {
|
|
if err == ErrNotFound {
|
|
return domain.ImportTask{}, false, nil
|
|
}
|
|
return domain.ImportTask{}, false, err
|
|
}
|
|
t, err := parseImportTask(raw)
|
|
return t, err == nil, err
|
|
}
|
|
|
|
// ── Parsers ───────────────────────────────────────────────────────────────────
|
|
|
|
func parseScrapeTask(raw json.RawMessage) (domain.ScrapeTask, error) {
|
|
var rec struct {
|
|
ID string `json:"id"`
|
|
Kind string `json:"kind"`
|
|
TargetURL string `json:"target_url"`
|
|
FromChapter int `json:"from_chapter"`
|
|
ToChapter int `json:"to_chapter"`
|
|
WorkerID string `json:"worker_id"`
|
|
Status string `json:"status"`
|
|
BooksFound int `json:"books_found"`
|
|
ChaptersScraped int `json:"chapters_scraped"`
|
|
ChaptersSkipped int `json:"chapters_skipped"`
|
|
Errors int `json:"errors"`
|
|
Started string `json:"started"`
|
|
Finished string `json:"finished"`
|
|
ErrorMessage string `json:"error_message"`
|
|
}
|
|
if err := json.Unmarshal(raw, &rec); err != nil {
|
|
return domain.ScrapeTask{}, err
|
|
}
|
|
started, _ := time.Parse(time.RFC3339, rec.Started)
|
|
finished, _ := time.Parse(time.RFC3339, rec.Finished)
|
|
return domain.ScrapeTask{
|
|
ID: rec.ID,
|
|
Kind: rec.Kind,
|
|
TargetURL: rec.TargetURL,
|
|
FromChapter: rec.FromChapter,
|
|
ToChapter: rec.ToChapter,
|
|
WorkerID: rec.WorkerID,
|
|
Status: domain.TaskStatus(rec.Status),
|
|
BooksFound: rec.BooksFound,
|
|
ChaptersScraped: rec.ChaptersScraped,
|
|
ChaptersSkipped: rec.ChaptersSkipped,
|
|
Errors: rec.Errors,
|
|
Started: started,
|
|
Finished: finished,
|
|
ErrorMessage: rec.ErrorMessage,
|
|
}, nil
|
|
}
|
|
|
|
func parseAudioTask(raw json.RawMessage) (domain.AudioTask, error) {
|
|
var rec struct {
|
|
ID string `json:"id"`
|
|
CacheKey string `json:"cache_key"`
|
|
Slug string `json:"slug"`
|
|
Chapter int `json:"chapter"`
|
|
Voice string `json:"voice"`
|
|
WorkerID string `json:"worker_id"`
|
|
Status string `json:"status"`
|
|
ErrorMessage string `json:"error_message"`
|
|
Started string `json:"started"`
|
|
Finished string `json:"finished"`
|
|
}
|
|
if err := json.Unmarshal(raw, &rec); err != nil {
|
|
return domain.AudioTask{}, err
|
|
}
|
|
started, _ := time.Parse(time.RFC3339, rec.Started)
|
|
finished, _ := time.Parse(time.RFC3339, rec.Finished)
|
|
return domain.AudioTask{
|
|
ID: rec.ID,
|
|
CacheKey: rec.CacheKey,
|
|
Slug: rec.Slug,
|
|
Chapter: rec.Chapter,
|
|
Voice: rec.Voice,
|
|
WorkerID: rec.WorkerID,
|
|
Status: domain.TaskStatus(rec.Status),
|
|
ErrorMessage: rec.ErrorMessage,
|
|
Started: started,
|
|
Finished: finished,
|
|
}, nil
|
|
}
|
|
|
|
func parseTranslationTask(raw json.RawMessage) (domain.TranslationTask, error) {
|
|
var rec struct {
|
|
ID string `json:"id"`
|
|
CacheKey string `json:"cache_key"`
|
|
Slug string `json:"slug"`
|
|
Chapter int `json:"chapter"`
|
|
Lang string `json:"lang"`
|
|
WorkerID string `json:"worker_id"`
|
|
Status string `json:"status"`
|
|
ErrorMessage string `json:"error_message"`
|
|
Started string `json:"started"`
|
|
Finished string `json:"finished"`
|
|
}
|
|
if err := json.Unmarshal(raw, &rec); err != nil {
|
|
return domain.TranslationTask{}, err
|
|
}
|
|
started, _ := time.Parse(time.RFC3339, rec.Started)
|
|
finished, _ := time.Parse(time.RFC3339, rec.Finished)
|
|
return domain.TranslationTask{
|
|
ID: rec.ID,
|
|
CacheKey: rec.CacheKey,
|
|
Slug: rec.Slug,
|
|
Chapter: rec.Chapter,
|
|
Lang: rec.Lang,
|
|
WorkerID: rec.WorkerID,
|
|
Status: domain.TaskStatus(rec.Status),
|
|
ErrorMessage: rec.ErrorMessage,
|
|
Started: started,
|
|
Finished: finished,
|
|
}, nil
|
|
}
|
|
|
|
func parseImportTask(raw json.RawMessage) (domain.ImportTask, error) {
|
|
var rec struct {
|
|
ID string `json:"id"`
|
|
Slug string `json:"slug"`
|
|
Title string `json:"title"`
|
|
FileName string `json:"file_name"`
|
|
FileType string `json:"file_type"`
|
|
ObjectKey string `json:"object_key"`
|
|
ChaptersKey string `json:"chapters_key"`
|
|
Author string `json:"author"`
|
|
CoverURL string `json:"cover_url"`
|
|
Genres string `json:"genres"` // stored as comma-separated
|
|
Summary string `json:"summary"`
|
|
BookStatus string `json:"book_status"`
|
|
WorkerID string `json:"worker_id"`
|
|
InitiatorUserID string `json:"initiator_user_id"`
|
|
Status string `json:"status"`
|
|
ChaptersDone int `json:"chapters_done"`
|
|
ChaptersTotal int `json:"chapters_total"`
|
|
ErrorMessage string `json:"error_message"`
|
|
Started string `json:"started"`
|
|
Finished string `json:"finished"`
|
|
}
|
|
if err := json.Unmarshal(raw, &rec); err != nil {
|
|
return domain.ImportTask{}, err
|
|
}
|
|
started, _ := time.Parse(time.RFC3339, rec.Started)
|
|
finished, _ := time.Parse(time.RFC3339, rec.Finished)
|
|
var genres []string
|
|
if rec.Genres != "" {
|
|
for _, g := range strings.Split(rec.Genres, ",") {
|
|
if g = strings.TrimSpace(g); g != "" {
|
|
genres = append(genres, g)
|
|
}
|
|
}
|
|
}
|
|
return domain.ImportTask{
|
|
ID: rec.ID,
|
|
Slug: rec.Slug,
|
|
Title: rec.Title,
|
|
FileName: rec.FileName,
|
|
FileType: rec.FileType,
|
|
ObjectKey: rec.ObjectKey,
|
|
ChaptersKey: rec.ChaptersKey,
|
|
Author: rec.Author,
|
|
CoverURL: rec.CoverURL,
|
|
Genres: genres,
|
|
Summary: rec.Summary,
|
|
BookStatus: rec.BookStatus,
|
|
WorkerID: rec.WorkerID,
|
|
InitiatorUserID: rec.InitiatorUserID,
|
|
Status: domain.TaskStatus(rec.Status),
|
|
ChaptersDone: rec.ChaptersDone,
|
|
ChaptersTotal: rec.ChaptersTotal,
|
|
ErrorMessage: rec.ErrorMessage,
|
|
Started: started,
|
|
Finished: finished,
|
|
}, nil
|
|
}
|
|
|
|
// ── CoverStore ─────────────────────────────────────────────────────────────────
|
|
|
|
func (s *Store) PutCover(ctx context.Context, slug string, data []byte, contentType string) error {
|
|
key := CoverObjectKey(slug)
|
|
if contentType == "" {
|
|
contentType = coverContentType(data)
|
|
}
|
|
if err := s.mc.putCover(ctx, key, contentType, data); err != nil {
|
|
return fmt.Errorf("PutCover: %w", err)
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func (s *Store) GetCover(ctx context.Context, slug string) ([]byte, string, bool, error) {
|
|
key := CoverObjectKey(slug)
|
|
data, ok, err := s.mc.getCover(ctx, key)
|
|
if err != nil {
|
|
return nil, "", false, fmt.Errorf("GetCover: %w", err)
|
|
}
|
|
if !ok {
|
|
return nil, "", false, nil
|
|
}
|
|
ct := coverContentType(data)
|
|
return data, ct, true, nil
|
|
}
|
|
|
|
// PutImportFile stores an uploaded import file (PDF/EPUB) in MinIO.
|
|
func (s *Store) PutImportFile(ctx context.Context, key string, data []byte) error {
|
|
return s.mc.putObject(ctx, "imports", key, "application/octet-stream", data)
|
|
}
|
|
|
|
// PutImportChapters stores a pre-parsed chapters JSON blob in MinIO.
|
|
func (s *Store) PutImportChapters(ctx context.Context, key string, data []byte) error {
|
|
return s.mc.putObject(ctx, "imports", key, "application/json", data)
|
|
}
|
|
|
|
// GetImportChapters retrieves the pre-parsed chapters JSON from MinIO.
|
|
func (s *Store) GetImportChapters(ctx context.Context, key string) ([]byte, error) {
|
|
data, err := s.mc.getObject(ctx, "imports", key)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("get chapters object: %w", err)
|
|
}
|
|
return data, nil
|
|
}
|
|
|
|
func (s *Store) CoverExists(ctx context.Context, slug string) bool {
|
|
return s.mc.coverExists(ctx, CoverObjectKey(slug))
|
|
}
|
|
|
|
// ── ChapterImageStore ──────────────────────────────────────────────────────────
|
|
|
|
func (s *Store) PutChapterImage(ctx context.Context, slug string, n int, data []byte, contentType string) error {
|
|
key := ChapterImageObjectKey(slug, n)
|
|
if contentType == "" {
|
|
contentType = coverContentType(data)
|
|
}
|
|
if err := s.mc.putChapterImage(ctx, key, contentType, data); err != nil {
|
|
return fmt.Errorf("PutChapterImage: %w", err)
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func (s *Store) GetChapterImage(ctx context.Context, slug string, n int) ([]byte, string, bool, error) {
|
|
key := ChapterImageObjectKey(slug, n)
|
|
data, ok, err := s.mc.getChapterImage(ctx, key)
|
|
if err != nil {
|
|
return nil, "", false, fmt.Errorf("GetChapterImage: %w", err)
|
|
}
|
|
if !ok {
|
|
return nil, "", false, nil
|
|
}
|
|
ct := coverContentType(data)
|
|
return data, ct, true, nil
|
|
}
|
|
|
|
func (s *Store) ChapterImageExists(ctx context.Context, slug string, n int) bool {
|
|
return s.mc.chapterImageExists(ctx, ChapterImageObjectKey(slug, n))
|
|
}
|
|
|
|
// ── TranslationStore ───────────────────────────────────────────────────────────
|
|
|
|
func (s *Store) TranslationObjectKey(lang, slug string, n int) string {
|
|
return TranslationObjectKey(lang, slug, n)
|
|
}
|
|
|
|
func (s *Store) TranslationExists(ctx context.Context, key string) bool {
|
|
return s.mc.objectExists(ctx, s.mc.bucketTranslations, key)
|
|
}
|
|
|
|
func (s *Store) PutTranslation(ctx context.Context, key string, data []byte) error {
|
|
return s.mc.putObject(ctx, s.mc.bucketTranslations, key, "text/markdown; charset=utf-8", data)
|
|
}
|
|
|
|
func (s *Store) GetTranslation(ctx context.Context, key string) (string, error) {
|
|
data, err := s.mc.getObject(ctx, s.mc.bucketTranslations, key)
|
|
if err != nil {
|
|
return "", fmt.Errorf("GetTranslation: %w", err)
|
|
}
|
|
return string(data), nil
|
|
}
|
|
|
|
// ── AIJobStore ────────────────────────────────────────────────────────────────
|
|
|
|
func (s *Store) CreateAIJob(ctx context.Context, job domain.AIJob) (string, error) {
|
|
payload := map[string]any{
|
|
"kind": job.Kind,
|
|
"slug": job.Slug,
|
|
"status": string(job.Status),
|
|
"from_item": job.FromItem,
|
|
"to_item": job.ToItem,
|
|
"items_done": job.ItemsDone,
|
|
"items_total": job.ItemsTotal,
|
|
"model": job.Model,
|
|
"payload": job.Payload,
|
|
"started": job.Started.Format(time.RFC3339),
|
|
}
|
|
var out struct {
|
|
ID string `json:"id"`
|
|
}
|
|
if err := s.pb.post(ctx, "/api/collections/ai_jobs/records", payload, &out); err != nil {
|
|
return "", fmt.Errorf("CreateAIJob: %w", err)
|
|
}
|
|
return out.ID, nil
|
|
}
|
|
|
|
func (s *Store) GetAIJob(ctx context.Context, id string) (domain.AIJob, bool, error) {
|
|
var raw json.RawMessage
|
|
if err := s.pb.get(ctx, fmt.Sprintf("/api/collections/ai_jobs/records/%s", id), &raw); err != nil {
|
|
if strings.Contains(err.Error(), "404") {
|
|
return domain.AIJob{}, false, nil
|
|
}
|
|
return domain.AIJob{}, false, fmt.Errorf("GetAIJob: %w", err)
|
|
}
|
|
job, err := parseAIJob(raw)
|
|
if err != nil {
|
|
return domain.AIJob{}, false, err
|
|
}
|
|
return job, true, nil
|
|
}
|
|
|
|
func (s *Store) UpdateAIJob(ctx context.Context, id string, fields map[string]any) error {
|
|
return s.pb.patch(ctx, fmt.Sprintf("/api/collections/ai_jobs/records/%s", id), fields)
|
|
}
|
|
|
|
func (s *Store) ListAIJobs(ctx context.Context) ([]domain.AIJob, error) {
|
|
items, err := s.pb.listAll(ctx, "ai_jobs", "", "-started")
|
|
if err != nil {
|
|
return nil, fmt.Errorf("ListAIJobs: %w", err)
|
|
}
|
|
out := make([]domain.AIJob, 0, len(items))
|
|
for _, raw := range items {
|
|
j, err := parseAIJob(raw)
|
|
if err != nil {
|
|
continue
|
|
}
|
|
out = append(out, j)
|
|
}
|
|
return out, nil
|
|
}
|
|
|
|
func parseAIJob(raw json.RawMessage) (domain.AIJob, error) {
|
|
var r struct {
|
|
ID string `json:"id"`
|
|
Kind string `json:"kind"`
|
|
Slug string `json:"slug"`
|
|
Status string `json:"status"`
|
|
FromItem int `json:"from_item"`
|
|
ToItem int `json:"to_item"`
|
|
ItemsDone int `json:"items_done"`
|
|
ItemsTotal int `json:"items_total"`
|
|
Model string `json:"model"`
|
|
Payload string `json:"payload"`
|
|
ErrorMessage string `json:"error_message"`
|
|
Started string `json:"started"`
|
|
Finished string `json:"finished"`
|
|
HeartbeatAt string `json:"heartbeat_at"`
|
|
}
|
|
if err := json.Unmarshal(raw, &r); err != nil {
|
|
return domain.AIJob{}, fmt.Errorf("parseAIJob: %w", err)
|
|
}
|
|
parseT := func(s string) time.Time {
|
|
if s == "" {
|
|
return time.Time{}
|
|
}
|
|
t, _ := time.Parse(time.RFC3339, s)
|
|
return t
|
|
}
|
|
return domain.AIJob{
|
|
ID: r.ID,
|
|
Kind: r.Kind,
|
|
Slug: r.Slug,
|
|
Status: domain.TaskStatus(r.Status),
|
|
FromItem: r.FromItem,
|
|
ToItem: r.ToItem,
|
|
ItemsDone: r.ItemsDone,
|
|
ItemsTotal: r.ItemsTotal,
|
|
Model: r.Model,
|
|
Payload: r.Payload,
|
|
ErrorMessage: r.ErrorMessage,
|
|
Started: parseT(r.Started),
|
|
Finished: parseT(r.Finished),
|
|
HeartbeatAt: parseT(r.HeartbeatAt),
|
|
}, nil
|
|
}
|
|
|
|
// ── Push subscriptions ────────────────────────────────────────────────────────
|
|
|
|
// PushSubscription holds the Web Push subscription data for a single browser.
|
|
type PushSubscription struct {
|
|
ID string
|
|
UserID string
|
|
Endpoint string
|
|
P256DH string
|
|
Auth string
|
|
}
|
|
|
|
// SavePushSubscription upserts a Web Push subscription for a user.
|
|
// If a record with the same endpoint already exists it is updated in place.
|
|
func (s *Store) SavePushSubscription(ctx context.Context, sub PushSubscription) error {
|
|
filter := fmt.Sprintf("endpoint=%q", sub.Endpoint)
|
|
existing, err := s.pb.listAll(ctx, "push_subscriptions", filter, "")
|
|
if err != nil {
|
|
return fmt.Errorf("SavePushSubscription list: %w", err)
|
|
}
|
|
payload := map[string]any{
|
|
"user_id": sub.UserID,
|
|
"endpoint": sub.Endpoint,
|
|
"p256dh": sub.P256DH,
|
|
"auth": sub.Auth,
|
|
}
|
|
if len(existing) > 0 {
|
|
var rec struct {
|
|
ID string `json:"id"`
|
|
}
|
|
if json.Unmarshal(existing[0], &rec) == nil && rec.ID != "" {
|
|
return s.pb.patch(ctx, fmt.Sprintf("/api/collections/push_subscriptions/records/%s", rec.ID), payload)
|
|
}
|
|
}
|
|
return s.pb.post(ctx, "/api/collections/push_subscriptions/records", payload, nil)
|
|
}
|
|
|
|
// DeletePushSubscription removes a Web Push subscription by endpoint.
|
|
func (s *Store) DeletePushSubscription(ctx context.Context, userID, endpoint string) error {
|
|
filter := fmt.Sprintf("user_id=%q&&endpoint=%q", userID, endpoint)
|
|
items, err := s.pb.listAll(ctx, "push_subscriptions", filter, "")
|
|
if err != nil {
|
|
return fmt.Errorf("DeletePushSubscription list: %w", err)
|
|
}
|
|
for _, raw := range items {
|
|
var rec struct {
|
|
ID string `json:"id"`
|
|
}
|
|
if json.Unmarshal(raw, &rec) == nil && rec.ID != "" {
|
|
_ = s.pb.delete(ctx, fmt.Sprintf("/api/collections/push_subscriptions/records/%s", rec.ID))
|
|
}
|
|
}
|
|
return nil
|
|
}
|
|
|
|
// ListPushSubscriptionsByBook returns all push subscriptions belonging to users
|
|
// who have the given book slug in their library (user_library collection).
|
|
func (s *Store) ListPushSubscriptionsByBook(ctx context.Context, slug string) ([]PushSubscription, error) {
|
|
// Find all users who have this book in their library
|
|
libFilter := fmt.Sprintf("slug=%q&&user_id!=''", slug)
|
|
libItems, err := s.pb.listAll(ctx, "user_library", libFilter, "")
|
|
if err != nil {
|
|
return nil, fmt.Errorf("ListPushSubscriptionsByBook list library: %w", err)
|
|
}
|
|
|
|
// Collect unique user IDs
|
|
seen := make(map[string]bool)
|
|
var userIDs []string
|
|
for _, raw := range libItems {
|
|
var rec struct {
|
|
UserID string `json:"user_id"`
|
|
}
|
|
if json.Unmarshal(raw, &rec) == nil && rec.UserID != "" && !seen[rec.UserID] {
|
|
seen[rec.UserID] = true
|
|
userIDs = append(userIDs, rec.UserID)
|
|
}
|
|
}
|
|
if len(userIDs) == 0 {
|
|
return nil, nil
|
|
}
|
|
|
|
// Build OR filter for push_subscriptions
|
|
parts := make([]string, len(userIDs))
|
|
for i, uid := range userIDs {
|
|
parts[i] = fmt.Sprintf("user_id=%q", uid)
|
|
}
|
|
subFilter := strings.Join(parts, "||")
|
|
subItems, err := s.pb.listAll(ctx, "push_subscriptions", subFilter, "")
|
|
if err != nil {
|
|
return nil, fmt.Errorf("ListPushSubscriptionsByBook list subs: %w", err)
|
|
}
|
|
|
|
subs := make([]PushSubscription, 0, len(subItems))
|
|
for _, raw := range subItems {
|
|
var rec struct {
|
|
ID string `json:"id"`
|
|
UserID string `json:"user_id"`
|
|
Endpoint string `json:"endpoint"`
|
|
P256DH string `json:"p256dh"`
|
|
Auth string `json:"auth"`
|
|
}
|
|
if json.Unmarshal(raw, &rec) == nil && rec.Endpoint != "" {
|
|
subs = append(subs, PushSubscription{
|
|
ID: rec.ID,
|
|
UserID: rec.UserID,
|
|
Endpoint: rec.Endpoint,
|
|
P256DH: rec.P256DH,
|
|
Auth: rec.Auth,
|
|
})
|
|
}
|
|
}
|
|
return subs, nil
|
|
}
|
|
|
|
// NotifyUsersWithBook creates an in-app notification for every logged-in user
|
|
// who has slug in their library. Errors for individual users are logged but
|
|
// do not abort the loop. Returns the number of notifications created.
|
|
func (s *Store) NotifyUsersWithBook(ctx context.Context, slug, title, message, link string) int {
|
|
userIDs, err := s.ListUserIDsWithBook(ctx, slug)
|
|
if err != nil || len(userIDs) == 0 {
|
|
return 0
|
|
}
|
|
var n int
|
|
for _, uid := range userIDs {
|
|
if createErr := s.CreateNotification(ctx, uid, title, message, link); createErr == nil {
|
|
n++
|
|
}
|
|
}
|
|
return n
|
|
}
|
|
// who have slug in their user_library. Used to fan-out new-chapter notifications.
|
|
// Admin users and users who have opted out of in-app new-chapter notifications
|
|
// (notify_new_chapters=false on app_users) are excluded.
|
|
func (s *Store) ListUserIDsWithBook(ctx context.Context, slug string) ([]string, error) {
|
|
// Collect user IDs to skip: admins + opted-out users.
|
|
skipIDs := make(map[string]bool)
|
|
excludedItems, err := s.pb.listAll(ctx, "app_users", `role="admin"||notify_new_chapters=false`, "")
|
|
if err == nil {
|
|
for _, raw := range excludedItems {
|
|
var rec struct {
|
|
ID string `json:"id"`
|
|
}
|
|
if json.Unmarshal(raw, &rec) == nil && rec.ID != "" {
|
|
skipIDs[rec.ID] = true
|
|
}
|
|
}
|
|
}
|
|
|
|
filter := fmt.Sprintf("slug=%q&&user_id!=''", slug)
|
|
items, err := s.pb.listAll(ctx, "user_library", filter, "")
|
|
if err != nil {
|
|
return nil, fmt.Errorf("ListUserIDsWithBook: %w", err)
|
|
}
|
|
seen := make(map[string]bool)
|
|
var ids []string
|
|
for _, raw := range items {
|
|
var rec struct {
|
|
UserID string `json:"user_id"`
|
|
}
|
|
if json.Unmarshal(raw, &rec) == nil && rec.UserID != "" && !seen[rec.UserID] && !skipIDs[rec.UserID] {
|
|
seen[rec.UserID] = true
|
|
ids = append(ids, rec.UserID)
|
|
}
|
|
}
|
|
return ids, nil
|
|
}
|