// Package storage — PocketBase REST client. // // Collections expected in PocketBase: // // books — slug(text,unique), title, author, cover, status, genres(json), // summary, total_chapters(number), source_url, ranking(number), updated(date) // chapters_idx — slug(text), number(number), title, date_label, updated(date) // ranking — rank(number), slug(text,unique), title, author, cover, status, // genres(json), source_url, updated(date) // progress — session_id(text), slug(text), chapter(number), updated(date) // audio_cache — cache_key(text,unique), filename(text), updated(date) // app_users — username(text,unique), password_hash(text), role(text), created(date) // scraping_tasks — id(auto), kind(text), target_url(text), status(text), // books_found(number), chapters_scraped(number), // chapters_skipped(number), errors(number), // started(date), finished(date), error_message(text) // user_sessions — user_id(text), session_id(text,unique), user_agent(text), // ip(text), created_at(date), last_seen(date) package storage import ( "bytes" "context" "encoding/json" "fmt" "io" "log/slog" "net/http" "net/url" "strings" "sync" "time" "github.com/libnovel/scraper/internal/scraper" ) // PocketBaseConfig holds PocketBase connection settings. type PocketBaseConfig struct { BaseURL string // e.g. "http://pocketbase:8090" AdminEmail string AdminPassword string } // pbClient is a minimal PocketBase admin REST client. type pbClient struct { cfg PocketBaseConfig httpClient *http.Client log *slog.Logger tokenMu sync.RWMutex token string tokenExp time.Time } // newPBClient creates a new PocketBase client. It does not authenticate yet; // authentication happens lazily on the first API call. func newPBClient(cfg PocketBaseConfig, log *slog.Logger) *pbClient { return &pbClient{ cfg: cfg, httpClient: &http.Client{Timeout: 15 * time.Second}, log: log, } } // ─── Auth ───────────────────────────────────────────────────────────────────── func (p *pbClient) authenticate(ctx context.Context) error { body, _ := json.Marshal(map[string]string{ "identity": p.cfg.AdminEmail, "password": p.cfg.AdminPassword, }) req, err := http.NewRequestWithContext(ctx, http.MethodPost, p.cfg.BaseURL+"/api/collections/_superusers/auth-with-password", bytes.NewReader(body)) if err != nil { return err } req.Header.Set("Content-Type", "application/json") resp, err := p.httpClient.Do(req) if err != nil { return fmt.Errorf("pocketbase: auth: %w", err) } defer resp.Body.Close() if resp.StatusCode != http.StatusOK { b, _ := io.ReadAll(resp.Body) return fmt.Errorf("pocketbase: auth status %d: %s", resp.StatusCode, b) } var result struct { Token string `json:"token"` } if err := json.NewDecoder(resp.Body).Decode(&result); err != nil { return fmt.Errorf("pocketbase: decode auth: %w", err) } p.tokenMu.Lock() p.token = result.Token p.tokenExp = time.Now().Add(12 * time.Hour) p.tokenMu.Unlock() return nil } func (p *pbClient) authToken(ctx context.Context) (string, error) { p.tokenMu.RLock() tok, exp := p.token, p.tokenExp p.tokenMu.RUnlock() if tok != "" && time.Now().Before(exp) { return tok, nil } if err := p.authenticate(ctx); err != nil { return "", err } p.tokenMu.RLock() defer p.tokenMu.RUnlock() return p.token, nil } // ─── Generic CRUD helpers ────────────────────────────────────────────────────── func (p *pbClient) do(ctx context.Context, method, path string, body interface{}) (*http.Response, error) { tok, err := p.authToken(ctx) if err != nil { return nil, err } var bodyReader io.Reader if body != nil { b, _ := json.Marshal(body) bodyReader = bytes.NewReader(b) } req, err := http.NewRequestWithContext(ctx, method, p.cfg.BaseURL+path, bodyReader) if err != nil { return nil, err } req.Header.Set("Authorization", "Bearer "+tok) if body != nil { req.Header.Set("Content-Type", "application/json") } return p.httpClient.Do(req) } // listOne fetches the first matching record from a collection. func (p *pbClient) listOne(ctx context.Context, collection, filter string) (map[string]interface{}, error) { q := url.Values{} q.Set("filter", filter) q.Set("perPage", "1") path := fmt.Sprintf("/api/collections/%s/records?%s", collection, q.Encode()) resp, err := p.do(ctx, http.MethodGet, path, nil) if err != nil { return nil, err } defer resp.Body.Close() if resp.StatusCode == http.StatusNotFound { return nil, nil } if resp.StatusCode != http.StatusOK { b, _ := io.ReadAll(resp.Body) return nil, fmt.Errorf("pocketbase: listOne %s: status %d: %s", collection, resp.StatusCode, b) } var result struct { Items []map[string]interface{} `json:"items"` } if err := json.NewDecoder(resp.Body).Decode(&result); err != nil { return nil, fmt.Errorf("pocketbase: listOne %s: decode: %w", collection, err) } if len(result.Items) == 0 { return nil, nil } return result.Items[0], nil } // listAll returns all records from a collection matching filter by paginating // through all pages (PocketBase default page size is capped at 500). func (p *pbClient) listAll(ctx context.Context, collection, filter, sort string) ([]map[string]interface{}, error) { const perPage = 500 var all []map[string]interface{} for page := 1; ; page++ { q := url.Values{} if filter != "" { q.Set("filter", filter) } if sort != "" { q.Set("sort", sort) } q.Set("perPage", fmt.Sprintf("%d", perPage)) q.Set("page", fmt.Sprintf("%d", page)) path := fmt.Sprintf("/api/collections/%s/records?%s", collection, q.Encode()) resp, err := p.do(ctx, http.MethodGet, path, nil) if err != nil { return nil, err } if resp.StatusCode != http.StatusOK { b, _ := io.ReadAll(resp.Body) resp.Body.Close() return nil, fmt.Errorf("pocketbase: listAll %s: status %d: %s", collection, resp.StatusCode, b) } var result struct { Page int `json:"page"` TotalPages int `json:"totalPages"` Items []map[string]interface{} `json:"items"` } if err := json.NewDecoder(resp.Body).Decode(&result); err != nil { resp.Body.Close() return nil, fmt.Errorf("pocketbase: listAll %s: decode: %w", collection, err) } resp.Body.Close() all = append(all, result.Items...) if page >= result.TotalPages || len(result.Items) == 0 { break } } return all, nil } // upsert creates a record; if one matching filter already exists it updates it. func (p *pbClient) upsert(ctx context.Context, collection, filter string, data map[string]interface{}) error { existing, err := p.listOne(ctx, collection, filter) if err != nil { return err } if existing != nil { id := existing["id"].(string) resp, err := p.do(ctx, http.MethodPatch, fmt.Sprintf("/api/collections/%s/records/%s", collection, id), data) if err != nil { return err } defer resp.Body.Close() if resp.StatusCode != http.StatusOK { b, _ := io.ReadAll(resp.Body) return fmt.Errorf("pocketbase: upsert (patch) %s id=%s: status %d: %s", collection, id, resp.StatusCode, b) } return nil } resp, err := p.do(ctx, http.MethodPost, fmt.Sprintf("/api/collections/%s/records", collection), data) if err != nil { return err } defer resp.Body.Close() if resp.StatusCode != http.StatusOK && resp.StatusCode != http.StatusCreated { b, _ := io.ReadAll(resp.Body) return fmt.Errorf("pocketbase: upsert (create) %s: status %d: %s", collection, resp.StatusCode, b) } return nil } // deleteWhere deletes all records matching filter in collection. func (p *pbClient) deleteWhere(ctx context.Context, collection, filter string) error { items, err := p.listAll(ctx, collection, filter, "") if err != nil { return err } for _, item := range items { id, _ := item["id"].(string) resp, err := p.do(ctx, http.MethodDelete, fmt.Sprintf("/api/collections/%s/records/%s", collection, id), nil) if err != nil { return err } if resp.StatusCode != http.StatusNoContent && resp.StatusCode != http.StatusOK { b, _ := io.ReadAll(resp.Body) resp.Body.Close() return fmt.Errorf("pocketbase: deleteWhere %s id=%s: status %d: %s", collection, id, resp.StatusCode, b) } resp.Body.Close() } return nil } // ─── PocketBaseStore ────────────────────────────────────────────────────────── // PocketBaseStore implements the structured-data portion of the Store interface // backed by PocketBase REST API. type PocketBaseStore struct { pb *pbClient log *slog.Logger } // NewPocketBaseStore returns a connected PocketBaseStore. func NewPocketBaseStore(cfg PocketBaseConfig, log *slog.Logger) *PocketBaseStore { return &PocketBaseStore{pb: newPBClient(cfg, log), log: log} } // Ping verifies connectivity by authenticating. func (s *PocketBaseStore) Ping(ctx context.Context) error { _, err := s.pb.authToken(ctx) return err } // ─── Collections schema bootstrap ──────────────────────────────────────────── // CollectionDef maps a collection name to its fields for auto-creation. // EnsureCollections creates missing collections via the PocketBase API. // Safe to call on every startup — existing collections are skipped. func (s *PocketBaseStore) EnsureCollections(ctx context.Context) error { // We just attempt to create each collection; 400/422 errors for "already // exists" are silently ignored. // PocketBase v0.22+ uses "fields"; older versions used "schema". // We use "fields" which is the current API. collections := []map[string]interface{}{ { "name": "books", "type": "base", "fields": []map[string]interface{}{ {"name": "slug", "type": "text", "required": true}, {"name": "title", "type": "text", "required": true}, {"name": "author", "type": "text"}, {"name": "cover", "type": "text"}, {"name": "status", "type": "text"}, {"name": "genres", "type": "json"}, {"name": "summary", "type": "text"}, {"name": "total_chapters", "type": "number"}, {"name": "source_url", "type": "text"}, {"name": "ranking", "type": "number"}, {"name": "meta_updated", "type": "date"}, }, }, { "name": "chapters_idx", "type": "base", "fields": []map[string]interface{}{ {"name": "slug", "type": "text", "required": true}, {"name": "number", "type": "number", "required": true}, {"name": "title", "type": "text"}, {"name": "date_label", "type": "text"}, }, }, { "name": "ranking", "type": "base", "fields": []map[string]interface{}{ {"name": "rank", "type": "number", "required": true}, {"name": "slug", "type": "text", "required": true}, {"name": "title", "type": "text"}, {"name": "author", "type": "text"}, {"name": "cover", "type": "text"}, {"name": "status", "type": "text"}, {"name": "genres", "type": "json"}, {"name": "source_url", "type": "text"}, {"name": "updated", "type": "date"}, }, }, { "name": "progress", "type": "base", "fields": []map[string]interface{}{ {"name": "session_id", "type": "text", "required": true}, {"name": "user_id", "type": "text"}, {"name": "slug", "type": "text", "required": true}, {"name": "chapter", "type": "number"}, {"name": "updated", "type": "date"}, }, }, { "name": "audio_cache", "type": "base", "fields": []map[string]interface{}{ {"name": "cache_key", "type": "text", "required": true}, {"name": "filename", "type": "text"}, {"name": "updated", "type": "date"}, }, }, { "name": "app_users", "type": "base", "fields": []map[string]interface{}{ {"name": "username", "type": "text", "required": true}, {"name": "password_hash", "type": "text", "required": true}, {"name": "role", "type": "text"}, {"name": "created", "type": "date"}, }, }, { "name": "user_library", "type": "base", "fields": []map[string]interface{}{ {"name": "session_id", "type": "text", "required": true}, {"name": "user_id", "type": "text"}, {"name": "slug", "type": "text", "required": true}, {"name": "saved_at", "type": "date"}, }, }, { "name": "scraping_tasks", "type": "base", "fields": []map[string]interface{}{ {"name": "kind", "type": "text", "required": true}, // "catalogue" | "book" {"name": "target_url", "type": "text"}, // set for single-book scrapes {"name": "status", "type": "text", "required": true}, // "running" | "done" | "failed" | "cancelled" {"name": "books_found", "type": "number"}, {"name": "chapters_scraped", "type": "number"}, {"name": "chapters_skipped", "type": "number"}, {"name": "errors", "type": "number"}, {"name": "started", "type": "date"}, {"name": "finished", "type": "date"}, {"name": "error_message", "type": "text"}, }, }, { "name": "audio_jobs", "type": "base", "fields": []map[string]interface{}{ {"name": "cache_key", "type": "text", "required": true}, // "slug/chapter/voice" {"name": "slug", "type": "text", "required": true}, {"name": "chapter", "type": "number"}, {"name": "voice", "type": "text"}, {"name": "status", "type": "text", "required": true}, // "pending" | "generating" | "done" | "failed" {"name": "error_message", "type": "text"}, {"name": "started", "type": "date"}, {"name": "finished", "type": "date"}, }, }, { "name": "user_sessions", "type": "base", "fields": []map[string]interface{}{ {"name": "user_id", "type": "text", "required": true}, {"name": "session_id", "type": "text", "required": true}, // random ID embedded in auth token {"name": "user_agent", "type": "text"}, {"name": "ip", "type": "text"}, {"name": "created_at", "type": "date"}, {"name": "last_seen", "type": "date"}, }, }, } for _, col := range collections { name, _ := col["name"].(string) resp, err := s.pb.do(ctx, http.MethodPost, "/api/collections", col) if err != nil { return fmt.Errorf("pocketbase: ensure collection %q: %w", name, err) } b, _ := io.ReadAll(resp.Body) resp.Body.Close() switch resp.StatusCode { case http.StatusOK, http.StatusCreated: s.log.Info("pocketbase: collection created", "collection", name) case http.StatusBadRequest, http.StatusUnprocessableEntity: // Already exists or schema mismatch — expected on subsequent startups. s.log.Debug("pocketbase: collection already exists (skipped)", "collection", name) default: s.log.Warn("pocketbase: unexpected status ensuring collection", "collection", name, "status", resp.StatusCode, "body", string(b)) } } return nil } // ─── Schema migrations ──────────────────────────────────────────────────────── // migration describes a single field to guarantee exists in a collection. type migration struct { collection string fieldName string fieldType string } // migrations is the ordered list of schema changes applied on every startup. var migrations = []migration{ // user_id was added to progress after initial deploy. {"progress", "user_id", "text"}, } // EnsureMigrations idempotently adds any fields that are missing from existing // collections. It fetches the current schema, checks for each field by name, // and PATCHes the collection only when something is absent. // Safe to call on every startup — no-ops when schema is already up to date. func (s *PocketBaseStore) EnsureMigrations(ctx context.Context) error { for _, m := range migrations { if err := s.ensureField(ctx, m); err != nil { return err } } return nil } func (s *PocketBaseStore) ensureField(ctx context.Context, m migration) error { // Fetch current collection schema. resp, err := s.pb.do(ctx, http.MethodGet, "/api/collections/"+m.collection, nil) if err != nil { return fmt.Errorf("pocketbase: ensureField %s.%s: fetch schema: %w", m.collection, m.fieldName, err) } defer resp.Body.Close() body, _ := io.ReadAll(resp.Body) if resp.StatusCode != http.StatusOK { return fmt.Errorf("pocketbase: ensureField %s.%s: fetch schema status %d: %s", m.collection, m.fieldName, resp.StatusCode, body) } var schema struct { ID string `json:"id"` Fields []map[string]interface{} `json:"fields"` } if err := json.Unmarshal(body, &schema); err != nil { return fmt.Errorf("pocketbase: ensureField %s.%s: decode schema: %w", m.collection, m.fieldName, err) } // Check if field already exists. for _, f := range schema.Fields { if name, _ := f["name"].(string); name == m.fieldName { s.log.Debug("pocketbase: field already exists, skipping migration", "collection", m.collection, "field", m.fieldName) return nil } } // Append the new field and PATCH the collection. newFields := append(schema.Fields, map[string]interface{}{ "name": m.fieldName, "type": m.fieldType, }) patch := map[string]interface{}{"fields": newFields} patchResp, err := s.pb.do(ctx, http.MethodPatch, "/api/collections/"+schema.ID, patch) if err != nil { return fmt.Errorf("pocketbase: ensureField %s.%s: patch: %w", m.collection, m.fieldName, err) } defer patchResp.Body.Close() patchBody, _ := io.ReadAll(patchResp.Body) if patchResp.StatusCode != http.StatusOK { return fmt.Errorf("pocketbase: ensureField %s.%s: patch status %d: %s", m.collection, m.fieldName, patchResp.StatusCode, patchBody) } s.log.Info("pocketbase: schema migration applied", "collection", m.collection, "field", m.fieldName, "type", m.fieldType) return nil } // ─── Book metadata ──────────────────────────────────────────────────────────── func (s *PocketBaseStore) UpsertBook(ctx context.Context, slug, title, author, cover, status, summary, sourceURL string, genres []string, totalChapters, ranking int) error { genresJSON, _ := json.Marshal(genres) return s.pb.upsert(ctx, "books", fmt.Sprintf(`slug="%s"`, pbEsc(slug)), map[string]interface{}{ "slug": slug, "title": title, "author": author, "cover": cover, "status": status, "genres": string(genresJSON), "summary": summary, "total_chapters": totalChapters, "source_url": sourceURL, "ranking": ranking, "meta_updated": time.Now().UTC().Format(time.RFC3339), }) } func (s *PocketBaseStore) GetBook(ctx context.Context, slug string) (map[string]interface{}, bool, error) { rec, err := s.pb.listOne(ctx, "books", fmt.Sprintf(`slug="%s"`, pbEsc(slug))) if err != nil { return nil, false, err } if rec == nil { return nil, false, nil } return rec, true, nil } func (s *PocketBaseStore) ListBooks(ctx context.Context) ([]map[string]interface{}, error) { return s.pb.listAll(ctx, "books", "", "+title") } func (s *PocketBaseStore) BookMetaUpdated(ctx context.Context, slug string) (time.Time, error) { rec, err := s.pb.listOne(ctx, "books", fmt.Sprintf(`slug="%s"`, pbEsc(slug))) if err != nil || rec == nil { return time.Time{}, err } if ts, ok := rec["meta_updated"].(string); ok { t, err := time.Parse(time.RFC3339, ts) if err == nil { return t, nil } } return time.Time{}, nil } // ─── Chapter index ──────────────────────────────────────────────────────────── func (s *PocketBaseStore) UpsertChapterIdx(ctx context.Context, slug string, number int, title, dateLabel string) error { return s.pb.upsert(ctx, "chapters_idx", fmt.Sprintf(`slug="%s"&&number=%d`, pbEsc(slug), number), map[string]interface{}{ "slug": slug, "number": number, "title": title, "date_label": dateLabel, }) } // WriteChapterRefs upserts chapter index rows (number + title) for all refs // without writing any chapter text. Errors are logged and skipped; the // operation is best-effort. func (s *PocketBaseStore) WriteChapterRefs(ctx context.Context, slug string, refs []scraper.ChapterRef) error { var firstErr error for _, ref := range refs { if err := s.UpsertChapterIdx(ctx, slug, ref.Number, ref.Title, ""); err != nil { s.log.Warn("pocketbase: WriteChapterRefs: upsert failed", "slug", slug, "chapter", ref.Number, "err", err) if firstErr == nil { firstErr = err } } } return firstErr } func (s *PocketBaseStore) ListChapterIdx(ctx context.Context, slug string) ([]map[string]interface{}, error) { return s.pb.listAll(ctx, "chapters_idx", fmt.Sprintf(`slug="%s"`, pbEsc(slug)), "+number") } func (s *PocketBaseStore) CountChapterIdx(ctx context.Context, slug string) int { rows, err := s.ListChapterIdx(ctx, slug) if err != nil { s.log.Warn("pocketbase: CountChapterIdx failed", "slug", slug, "err", err) return 0 } return len(rows) } // ─── Ranking (per-item) ─────────────────────────────────────────────────────── func (s *PocketBaseStore) UpsertRankingItem(ctx context.Context, item RankingItem) error { genresJSON, _ := json.Marshal(item.Genres) return s.pb.upsert(ctx, "ranking", fmt.Sprintf(`slug="%s"`, pbEsc(item.Slug)), map[string]interface{}{ "rank": item.Rank, "slug": item.Slug, "title": item.Title, "author": item.Author, "cover": item.Cover, "status": item.Status, "genres": string(genresJSON), "source_url": item.SourceURL, "updated": time.Now().UTC().Format(time.RFC3339), }) } func (s *PocketBaseStore) ListRankingItems(ctx context.Context) ([]RankingItem, error) { rows, err := s.pb.listAll(ctx, "ranking", "", "+rank") if err != nil { return nil, err } items := make([]RankingItem, 0, len(rows)) for _, r := range rows { item := RankingItem{ Rank: int(floatVal(r, "rank")), Slug: strVal(r, "slug"), Title: strVal(r, "title"), Author: strVal(r, "author"), Cover: strVal(r, "cover"), Status: strVal(r, "status"), SourceURL: strVal(r, "source_url"), } if ts, ok := r["updated"].(string); ok { item.Updated, _ = time.Parse(time.RFC3339, ts) } switch v := r["genres"].(type) { case string: _ = json.Unmarshal([]byte(v), &item.Genres) case []interface{}: for _, g := range v { if s, ok := g.(string); ok { item.Genres = append(item.Genres, s) } } } items = append(items, item) } return items, nil } // RankingLastUpdated returns the most recent Updated time across all ranking rows, // or the zero time if no rows exist. func (s *PocketBaseStore) RankingLastUpdated(ctx context.Context) (time.Time, error) { // listAll with sort "-updated" and perPage=1 is the cheapest approach. q := url.Values{} q.Set("sort", "-updated") q.Set("perPage", "1") path := fmt.Sprintf("/api/collections/ranking/records?%s", q.Encode()) resp, err := s.pb.do(ctx, http.MethodGet, path, nil) if err != nil { return time.Time{}, err } defer resp.Body.Close() if resp.StatusCode != http.StatusOK { b, _ := io.ReadAll(resp.Body) return time.Time{}, fmt.Errorf("pocketbase: RankingLastUpdated: status %d: %s", resp.StatusCode, b) } var result struct { Items []map[string]interface{} `json:"items"` } if err := json.NewDecoder(resp.Body).Decode(&result); err != nil { return time.Time{}, fmt.Errorf("pocketbase: RankingLastUpdated: decode: %w", err) } if len(result.Items) == 0 { return time.Time{}, nil } ts, _ := result.Items[0]["updated"].(string) t, _ := time.Parse(time.RFC3339, ts) return t, nil } // ─── Reading progress ───────────────────────────────────────────────────────── func (s *PocketBaseStore) SetProgress(ctx context.Context, sessionID, slug string, chapter int) error { return s.pb.upsert(ctx, "progress", fmt.Sprintf(`session_id="%s"&&slug="%s"`, pbEsc(sessionID), pbEsc(slug)), map[string]interface{}{ "session_id": sessionID, "slug": slug, "chapter": chapter, "updated": time.Now().UTC().Format(time.RFC3339), }) } func (s *PocketBaseStore) GetProgress(ctx context.Context, sessionID, slug string) (int, time.Time, bool, error) { rec, err := s.pb.listOne(ctx, "progress", fmt.Sprintf(`session_id="%s"&&slug="%s"`, pbEsc(sessionID), pbEsc(slug))) if err != nil { return 0, time.Time{}, false, err } if rec == nil { return 0, time.Time{}, false, nil } ch := int(floatVal(rec, "chapter")) var updated time.Time if ts, ok := rec["updated"].(string); ok { updated, _ = time.Parse(time.RFC3339, ts) } return ch, updated, true, nil } func (s *PocketBaseStore) AllProgress(ctx context.Context, sessionID string) ([]map[string]interface{}, error) { return s.pb.listAll(ctx, "progress", fmt.Sprintf(`session_id="%s"`, pbEsc(sessionID)), "-updated") } func (s *PocketBaseStore) DeleteProgress(ctx context.Context, sessionID, slug string) error { return s.pb.deleteWhere(ctx, "progress", fmt.Sprintf(`session_id="%s"&&slug="%s"`, pbEsc(sessionID), pbEsc(slug))) } // ─── Audio cache ────────────────────────────────────────────────────────────── func (s *PocketBaseStore) SetAudioCache(ctx context.Context, cacheKey, filename string) error { return s.pb.upsert(ctx, "audio_cache", fmt.Sprintf(`cache_key="%s"`, pbEsc(cacheKey)), map[string]interface{}{ "cache_key": cacheKey, "filename": filename, "updated": time.Now().UTC().Format(time.RFC3339), }) } func (s *PocketBaseStore) GetAudioCache(ctx context.Context, cacheKey string) (string, bool, error) { rec, err := s.pb.listOne(ctx, "audio_cache", fmt.Sprintf(`cache_key="%s"`, pbEsc(cacheKey))) if err != nil { return "", false, err } if rec == nil { return "", false, nil } filename, _ := rec["filename"].(string) return filename, filename != "", nil } // ─── Scraping tasks ─────────────────────────────────────────────────────────── // CreateScrapingTask inserts a new scraping_tasks record with status="running" // and returns the newly created record's ID. func (s *PocketBaseStore) CreateScrapingTask(ctx context.Context, kind, targetURL string) (string, error) { data := map[string]interface{}{ "kind": kind, "target_url": targetURL, "status": "running", "books_found": 0, "chapters_scraped": 0, "chapters_skipped": 0, "errors": 0, "started": time.Now().UTC().Format(time.RFC3339), } resp, err := s.pb.do(ctx, http.MethodPost, "/api/collections/scraping_tasks/records", data) if err != nil { return "", err } defer resp.Body.Close() b, _ := io.ReadAll(resp.Body) if resp.StatusCode != http.StatusOK && resp.StatusCode != http.StatusCreated { return "", fmt.Errorf("pocketbase: CreateScrapingTask: status %d: %s", resp.StatusCode, b) } var rec map[string]interface{} if err := json.Unmarshal(b, &rec); err != nil { return "", fmt.Errorf("pocketbase: CreateScrapingTask: decode: %w", err) } id, _ := rec["id"].(string) return id, nil } // UpdateScrapingTask patches counters on an existing scraping_tasks record. func (s *PocketBaseStore) UpdateScrapingTask(ctx context.Context, id string, data map[string]interface{}) error { resp, err := s.pb.do(ctx, http.MethodPatch, fmt.Sprintf("/api/collections/scraping_tasks/records/%s", id), data) if err != nil { return err } defer resp.Body.Close() if resp.StatusCode != http.StatusOK { b, _ := io.ReadAll(resp.Body) return fmt.Errorf("pocketbase: UpdateScrapingTask id=%s: status %d: %s", id, resp.StatusCode, b) } return nil } // ListScrapingTasks returns all scraping_tasks sorted by started descending. func (s *PocketBaseStore) ListScrapingTasks(ctx context.Context) ([]map[string]interface{}, error) { return s.pb.listAll(ctx, "scraping_tasks", "", "-started") } // ─── Audio jobs ─────────────────────────────────────────────────────────────── // CreateAudioJob inserts a new audio_jobs record with status="pending". func (s *PocketBaseStore) CreateAudioJob(ctx context.Context, slug string, chapter int, voice string) (string, error) { cacheKey := fmt.Sprintf("%s/%d/%s", slug, chapter, voice) data := map[string]interface{}{ "cache_key": cacheKey, "slug": slug, "chapter": chapter, "voice": voice, "status": "pending", "started": time.Now().UTC().Format(time.RFC3339), } resp, err := s.pb.do(ctx, http.MethodPost, "/api/collections/audio_jobs/records", data) if err != nil { return "", err } defer resp.Body.Close() b, _ := io.ReadAll(resp.Body) if resp.StatusCode != http.StatusOK && resp.StatusCode != http.StatusCreated { return "", fmt.Errorf("pocketbase: CreateAudioJob: status %d: %s", resp.StatusCode, b) } var rec map[string]interface{} if err := json.Unmarshal(b, &rec); err != nil { return "", fmt.Errorf("pocketbase: CreateAudioJob: decode: %w", err) } id, _ := rec["id"].(string) return id, nil } // UpdateAudioJob patches status, error_message, and optionally finished on an audio_jobs record. func (s *PocketBaseStore) UpdateAudioJob(ctx context.Context, id, status, errMsg string, finished time.Time) error { data := map[string]interface{}{ "status": status, "error_message": errMsg, } if !finished.IsZero() { data["finished"] = finished.UTC().Format(time.RFC3339) } resp, err := s.pb.do(ctx, http.MethodPatch, fmt.Sprintf("/api/collections/audio_jobs/records/%s", id), data) if err != nil { return err } defer resp.Body.Close() if resp.StatusCode != http.StatusOK { b, _ := io.ReadAll(resp.Body) return fmt.Errorf("pocketbase: UpdateAudioJob id=%s: status %d: %s", id, resp.StatusCode, b) } return nil } // GetAudioJob returns the most recent audio_jobs record for the given cache key. func (s *PocketBaseStore) GetAudioJob(ctx context.Context, cacheKey string) (map[string]interface{}, bool, error) { rec, err := s.pb.listOne(ctx, "audio_jobs", fmt.Sprintf(`cache_key="%s"`, pbEsc(cacheKey))) if err != nil { return nil, false, err } if rec == nil { return nil, false, nil } return rec, true, nil } // ListAudioJobs returns all audio_jobs sorted by started descending. func (s *PocketBaseStore) ListAudioJobs(ctx context.Context) ([]map[string]interface{}, error) { return s.pb.listAll(ctx, "audio_jobs", "", "-started") } // ─── helpers ────────────────────────────────────────────────────────────────── // pbEsc escapes a string for use in a PocketBase filter expression. // Only escapes double-quotes to prevent injection. func pbEsc(s string) string { return strings.ReplaceAll(s, `"`, `\"`) } func floatVal(m map[string]interface{}, key string) float64 { if v, ok := m[key].(float64); ok { return v } return 0 }