libnovel/scraper/internal/storage/pocketbase.go

// 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)
package storage

import (
	"bytes"
	"context"
	"encoding/json"
	"fmt"
	"io"
	"log/slog"
	"net/http"
	"net/url"
	"strings"
	"sync"
	"time"
)

// 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 (up to 500) from a collection matching filter.
func (p *pbClient) listAll(ctx context.Context, collection, filter, sort string) ([]map[string]interface{}, error) {
	q := url.Values{}
	if filter != "" {
		q.Set("filter", filter)
	}
	if sort != "" {
		q.Set("sort", sort)
	}
	q.Set("perPage", "500")
	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.StatusOK {
		b, _ := io.ReadAll(resp.Body)
		return nil, fmt.Errorf("pocketbase: listAll %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: listAll %s: decode: %w", collection, err)
	}
	return result.Items, 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
		}
		defer resp.Body.Close()
		if resp.StatusCode != http.StatusNoContent && resp.StatusCode != http.StatusOK {
			b, _ := io.ReadAll(resp.Body)
			return fmt.Errorf("pocketbase: deleteWhere %s id=%s: status %d: %s", collection, id, resp.StatusCode, b)
		}
	}
	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": "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": "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"},
			},
		},
	}
	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
}

// ─── 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,
		})
}

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")
}

// ─── 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
}