Files
mangaBookmark/backend/internal/store/store.go
T
sulthan 2b597921f4 review: clear stale attribution, gate numberless PUTs, cite the ADR
A Poll finding a higher number now clears series.latest_raised_by: the value
it stores is its own, so the next Poll must not credit that Reader with an
agreement they did not earn, nor charge them for a retraction of a number
they never reported.

A PUT carrying no chapter is no Sighting and defers nothing. The ceiling now
counts the Site's own rest rather than defaultRest. The userscripts PUT an
unchanged read too - the case the whole mechanism exists for was the one they
never sent.

ADR-0011 carries its citations and states honestly when a Reader's marks take
effect.
2026-08-16 16:36:31 +07:00

1180 lines
48 KiB
Go

package store
import (
"crypto/sha256"
"database/sql"
"embed"
"encoding/hex"
"errors"
"fmt"
"io/fs"
"os"
"path"
"path/filepath"
"regexp"
"slices"
"strconv"
"strings"
_ "github.com/jackc/pgx/v5/stdlib"
)
// Bookmark is one tracked series, keyed "<site>:<series_id>" across both sites.
//
// LastChapter* is the user's read progress; LatestChapter* is the newest
// chapter the site has published, captured opportunistically by the userscript.
//
// Title, SeriesURL, Cover, Kind and LatestChapter* live on the shared Series
// row (ADR-0003) and are joined in on read; Bookmark carries only what differs
// between readers: progress, favourite, lifecycle bucket, updated_at. The wire
// format stays flat regardless — see ADR-0004.
type Bookmark struct {
Key string `json:"key"`
Site string `json:"site"`
SeriesID string `json:"series_id"`
Title string `json:"title"`
SeriesURL string `json:"series_url"`
// Cover is the wire value: an absolute URL on this deployment's own
// origin once the bytes exist, and "" until they do — never a third-party
// address and never an address that 404s (ADR-0007). A client may still
// send this field and it is discarded on the way in; see Upsert.
Cover string `json:"cover"`
LastChapter string `json:"last_chapter"`
LastChapterNum float64 `json:"last_chapter_num"`
LastChapterURL string `json:"last_chapter_url"`
Favorite bool `json:"favorite"`
LatestChapter string `json:"latest_chapter"`
LatestChapterNum *float64 `json:"latest_chapter_num"` // nil until first captured
UpdatedAt int64 `json:"updated_at"` // unix ms; see Upsert
// Status is the lifecycle bucket: reading, archived, or finished.
// Archived series stay polled for new chapters; finished ones do not.
// Empty on the way in means "no opinion" — see Upsert.
Status string `json:"status"`
// Kind is the library bucket: manga or novel. Empty on the way in means
// "no opinion" — see Upsert.
Kind string `json:"kind"`
}
// Series is one distinct work, shared by every bookmark that tracks it. It is
// keyed (site, series_id) — the pair a bookmark key decomposes into — and
// exists once no matter how many bookmarks point at it (ADR-0003).
//
// Title, SeriesURL and Cover are written once, at creation: a PUT naming an
// existing Series has them ignored, and only the backend's own Poll may change
// them. Kind and the latest-chapter fields are last-write-wins like the
// bookmark's own fields. Never serialized: the wire format is the flat
// Bookmark (ADR-0004).
type Series struct {
Site string
SeriesID string
Title string
SeriesURL string
// Cover is the third-party source address the bytes come from, and
// CoverAddress the content address they are stored under. A blank
// CoverAddress is what "no Cover yet" means: the poll fills it and never
// replaces a filled one (ADR-0007).
Cover string
CoverAddress string
Kind string
LatestChapter string
LatestChapterNum *float64 // nil until first captured
LatestCheckedAt int64 // unix ms; see MarkLatestChecked
// LatestRaisedBy is the Reader whose Sighting last raised LatestChapter,
// and nil when the stored value is a Poll's own finding. It is what lets a
// Poll that contradicts the value downwards name a Reader instead of
// merely flagging the row (issue #103); the Poll that judges it clears it.
LatestRaisedBy *int64
// readerCount is the number of bookmarks referencing this series, filled
// only by the due-queue query that orders on it.
readerCount int
}
// Key returns the canonical identity in bookmark-key form ("<site>:<series_id>"),
// used by the poller's logs and by tests asserting on the due queue.
func (s Series) Key() string { return s.Site + ":" + s.SeriesID }
// HasNewChapter reports whether the site has published past the read point.
// A nil LatestChapterNum means nothing has been captured yet, which is not the
// same as "nothing new".
func (b Bookmark) HasNewChapter() bool {
return b.LatestChapterNum != nil && *b.LatestChapterNum > b.LastChapterNum
}
// chapterLeadIn matches the prefix the userscript and the poller both write
// ("Chapter 250"), so the UI can add exactly one "Ch " of its own instead of
// doubling it. A manual edit through the web UI stores a bare "250", which is
// the same string minus the lead-in.
var chapterLeadIn = regexp.MustCompile(`(?i)^\s*(?:chapter|ch\.?)\s*`)
func displayChapter(raw string, num float64) string {
rest := strings.TrimSpace(chapterLeadIn.ReplaceAllString(raw, ""))
if rest == "" {
rest = strconv.FormatFloat(num, 'f', -1, 64)
}
// "Ch " only makes sense in front of a number; anything else is a label the
// site gave us, so pass it through as written.
if rest[0] < '0' || rest[0] > '9' {
return rest
}
return "Ch " + rest
}
// DisplayChapter is the read-progress line: one canonical "Ch N" whatever
// format the write came in as.
func (b Bookmark) DisplayChapter() string {
return displayChapter(b.LastChapter, b.LastChapterNum)
}
// DisplayLatest is the same for the newest published chapter, which arrives
// with the same "Chapter N" lead-in from both the userscript and the poller.
func (b Bookmark) DisplayLatest() string {
var num float64
if b.LatestChapterNum != nil {
num = *b.LatestChapterNum
}
return displayChapter(b.LatestChapter, num)
}
// ContinueURL is where the Continue button points: the chapter last read, or
// the series page when no chapter URL was ever captured.
func (b Bookmark) ContinueURL() string {
if b.LastChapterURL != "" {
return b.LastChapterURL
}
return b.SeriesURL
}
// Initial is the monogram the web UI shows in place of a cover when the
// source site never gave us an og:image. First rune, uppercased; "?" when even
// the title is missing, so the slot is never empty.
func (b Bookmark) Initial() string {
for _, r := range b.Title {
return strings.ToUpper(string(r))
}
return "?"
}
// CoverContentType canonicalises a fetched response's media type and reports
// whether the bytes are safe to store and serve. comix answers "image/jpg",
// which no standard lists but browsers accept; it is stored as the real name
// rather than passed through, so one image never lands under two spellings.
func CoverContentType(contentType string) (string, bool) {
switch contentType {
case "image/jpg":
return "image/jpeg", true
case "image/webp", "image/jpeg", "image/png", "image/avif", "image/gif":
return contentType, true
default:
return "", false
}
}
// Library buckets. A bookmark is in exactly one. This cannot be derived from
// Site: asurascans serves manga and novels from the same /comics/ path, so the
// userscript that recorded the page is the only party that knows which.
const (
KindManga = "manga"
KindNovel = "novel"
)
// Lifecycle buckets. A bookmark is in exactly one; favorite is orthogonal.
const (
StatusReading = "reading"
StatusArchived = "archived"
StatusFinished = "finished"
)
//go:embed migrations/*.sql
var migrations embed.FS
// bookmarkColumns is the only value ever concatenated into query text. It is a
// compile-time constant; every request value is bound as a parameter. The
// series-owned fields are joined in from the series table, in scanBookmark
// order, so the flat Bookmark reads back whole despite the split (ADR-0004).
const bookmarkColumns = `b.site, b.series_id, s.title, s.series_url, s.cover_address,
b.last_chapter, b.last_chapter_num, b.last_chapter_url,
b.favorite, s.latest_chapter, s.latest_chapter_num, b.updated_at, b.status, s.kind`
// seriesColumns is the series row in scanSeries order, used by the poller's
// due query. latest_checked_at lives only on series — see MarkLatestChecked
// for why it stays off every client-visible write.
const seriesColumns = `s.site, s.series_id, s.title, s.series_url, s.cover, s.cover_address,
s.kind, s.latest_chapter, s.latest_chapter_num, s.latest_checked_at, s.latest_raised_by`
// Owner is the person running the service: the first Reader, seeded at startup
// so a fresh deployment has a library before anyone logs in. The seed makes
// sure exactly one readers row matches their Discord ID, carrying the SHA-256
// of their epoch-0 userscript credential (derived by internal/token). Every
// other Reader is created by their own first login (EnsureReader).
type Owner struct {
DiscordID string
// TokenHash is the SHA-256 of the epoch-0 credential; the array shape
// makes it a compile error to store anything that is not a hash.
TokenHash [32]byte
}
// Store is the Postgres-backed bookmark store.
type Store struct {
db *sql.DB
// ownerID is the seeded owner Reader (issue #22) — the only Reader with
// administrative reach (revoking another Reader's sessions). Every store
// method takes a reader id explicitly, so ownership is never implicit.
ownerID int64
coverDir string
// coverBaseURL is this deployment's public origin. Cover addresses are
// absolute because the userscript renders them on third-party origins,
// where a relative path would resolve against the Site (ADR-0007).
coverBaseURL string
// OnSeriesCreated fires once, after commit, for a Series no Reader had
// bookmarked before. It is how creation-time Cover and Latest Chapter
// acquisition is triggered without the write waiting on a third-party
// Site; nil disables it, which is what every test that does not care
// about acquisition leaves it as.
OnSeriesCreated func(Series)
}
// OwnerID returns the seeded owner Reader's id: the administrator, and the
// Reader every pre-registration bookmark belongs to.
func (s *Store) OwnerID() int64 { return s.ownerID }
// ReaderIDForTokenHash resolves the Reader whose stored credential hash
// matches, reporting absence with ok=false. The comparison is an equality on
// the 32-byte SHA-256 of the presented credential — never on the credential
// itself — and the indexed lookup reveals only whether some Reader matches,
// which the 401/200 split has to reveal anyway. An attacker's probe is the
// hash of their guess, so even the index's prefix comparisons leak nothing
// about the real credential.
func (s *Store) ReaderIDForTokenHash(hash [32]byte) (int64, bool, error) {
var id int64
err := s.db.QueryRow(
`SELECT id FROM readers WHERE token_sha256 = $1`, hash[:]).Scan(&id)
if errors.Is(err, sql.ErrNoRows) {
return 0, false, nil
}
if err != nil {
return 0, false, fmt.Errorf("reader by token hash: %w", err)
}
return id, true, nil
}
// ReaderTokenInfo returns the identity halves a Reader's credential is
// derived from (internal/token.Token): their Discord id and token epoch. The
// web UI needs these to rebuild the install URL — the only place a credential
// is ever produced in plaintext.
func (s *Store) ReaderTokenInfo(readerID int64) (string, int64, error) {
var (
discordID string
epoch int64
)
err := s.db.QueryRow(
`SELECT discord_id, token_epoch FROM readers WHERE id = $1`, readerID).
Scan(&discordID, &epoch)
if err != nil {
return "", 0, fmt.Errorf("reader %d token info: %w", readerID, err)
}
return discordID, epoch, nil
}
// RotateToken bumps a Reader's token epoch and rewrites the stored hash in
// one statement, so the new hash always matches the new epoch. expectedEpoch
// is the epoch the caller derived newHash for (ReaderTokenInfo + 1); a
// concurrent rotation — or an unknown reader — leaves the row untouched and
// is reported as an error rather than silently succeeding.
func (s *Store) RotateToken(readerID, expectedEpoch int64, newHash [32]byte) error {
var epoch int64
err := s.db.QueryRow(`
UPDATE readers SET token_epoch = token_epoch + 1, token_sha256 = $3
WHERE id = $1 AND token_epoch = $2
RETURNING token_epoch`, readerID, expectedEpoch, newHash[:]).Scan(&epoch)
if errors.Is(err, sql.ErrNoRows) {
return fmt.Errorf("rotate token for reader %d: concurrent rotation or unknown reader", readerID)
}
if err != nil {
return fmt.Errorf("rotate token for reader %d: %w", readerID, err)
}
return nil
}
// EnsureReader returns the Reader registered to discordID, creating the row on
// first sight. Registration is open to every guild member (issue #27), and the
// Discord identity is the only thing that decides which Reader a login is: one
// code path serves the first login and every later one, so a returning Reader
// can never end up with a second library.
//
// epochZeroHash is only used for a brand-new row. An existing row keeps its
// stored hash untouched, or a login would silently undo a rotation and revive
// the credential the Reader rotated away from.
func (s *Store) EnsureReader(discordID string, epochZeroHash [32]byte) (int64, error) {
var id int64
// DO UPDATE rather than DO NOTHING because only an updated row is
// returned by RETURNING; assigning the column to itself is the no-op that
// makes the existing id come back.
err := s.db.QueryRow(`
INSERT INTO readers (discord_id, token_sha256) VALUES ($1, $2)
ON CONFLICT (discord_id) DO UPDATE SET discord_id = readers.discord_id
RETURNING id`, discordID, epochZeroHash[:]).Scan(&id)
if err != nil {
return 0, fmt.Errorf("ensure reader: %w", err)
}
return id, nil
}
// SightingDisagreementLimit is the number of contradictions that stop that
// Reader's Sightings from deferring a Poll (issue #103). The counters exist
// before the mechanism that moves them, so the admin page (issue #102) can
// clear a false mark without waiting for the Sighting feature.
const SightingDisagreementLimit = 3
// Blocked reports whether this Reader's Sighting marks have reached the
// disagreement limit, which stops their Sightings from deferring a Poll.
func (r ReaderSummary) Blocked() bool {
return r.Disagreements >= SightingDisagreementLimit
}
// ReaderSummary is one Reader as the owner's administration panel sees them:
// who they are, how many live sessions they hold, and their Sighting marks.
// No credential material, hashed or otherwise, is exposed.
type ReaderSummary struct {
ID int64
DiscordID string
// Sessions counts unexpired session rows — what the owner revokes.
Sessions int
// Agreements and Disagreements are the Sighting counters (issue #102);
// zero means a trusted Reader.
Agreements int
Disagreements int
}
// Readers lists every Reader with their live session count and Sighting
// marks, oldest first, so the owner row (always the oldest) heads the list.
func (s *Store) Readers() ([]ReaderSummary, error) {
rows, err := s.db.Query(`
SELECT r.id, r.discord_id,
r.sighting_agreements, r.sighting_disagreements,
count(sess.id) FILTER (WHERE sess.expires_at > now()) AS sessions
FROM readers r
LEFT JOIN sessions sess ON sess.reader_id = r.id
GROUP BY r.id, r.discord_id, r.sighting_agreements, r.sighting_disagreements
ORDER BY r.id`)
if err != nil {
return nil, fmt.Errorf("query readers: %w", err)
}
defer rows.Close()
out := []ReaderSummary{}
for rows.Next() {
var r ReaderSummary
if err := rows.Scan(&r.ID, &r.DiscordID, &r.Agreements, &r.Disagreements, &r.Sessions); err != nil {
return nil, fmt.Errorf("scan reader: %w", err)
}
out = append(out, r)
}
return out, rows.Err()
}
// ClearReaderMarks zeroes a Reader's Sighting counters. It is the owner's
// remedy for a mark produced by a broken Site adapter rather than a dishonest
// Reader: it restores a privilege, it is not destruction.
func (s *Store) ClearReaderMarks(readerID int64) error {
if _, err := s.db.Exec(`
UPDATE readers SET sighting_agreements = 0, sighting_disagreements = 0
WHERE id = $1`, readerID); err != nil {
return fmt.Errorf("clear reader marks for reader %d: %w", readerID, err)
}
return nil
}
// readersMigration is the version that creates the readers table. The owner
// seed runs between two migrate passes, so that the run-once migration which
// attaches existing bookmarks (0004) finds the owner row.
const readersMigration = 3
// allMigrations is the migrate() cap that applies every pending version.
const allMigrations = 0
// Open connects to Postgres at url — a libpq connection URL such as
// "postgres://user:pass@host:5432/bookmarks?sslmode=disable" — brings its
// schema up to date, seeds the owner Reader, and prepares cover storage.
func Open(url string, owner Owner, coverDir, coverBaseURL string) (*Store, error) {
if strings.TrimSpace(coverDir) == "" {
return nil, errors.New("cover directory is required")
}
// Every wire Cover is this string with a path glued on, rendered by a
// userscript on a Site's own origin: anything but an absolute origin
// produces addresses no client can load, silently (ADR-0007).
base := strings.TrimRight(coverBaseURL, "/")
if host, ok := strings.CutPrefix(base, "https://"); !ok || host == "" {
if host, ok := strings.CutPrefix(base, "http://"); !ok || host == "" {
return nil, fmt.Errorf("cover base URL %q is not an absolute http(s) origin", coverBaseURL)
}
}
if err := os.MkdirAll(coverDir, 0o755); err != nil {
return nil, fmt.Errorf("create cover directory: %w", err)
}
info, err := os.Stat(coverDir)
if err != nil {
return nil, fmt.Errorf("stat cover directory: %w", err)
}
if !info.IsDir() {
return nil, fmt.Errorf("cover directory %q is not a directory", coverDir)
}
db, err := sql.Open("pgx", url)
if err != nil {
return nil, fmt.Errorf("open postgres: %w", err)
}
// Schema runs in two passes with the seed between: 0003 creates the
// readers table, the owner row must exist before 0004 attaches the
// existing bookmarks to it. Anything past 0004 is applied by the second
// pass.
if err := migrate(db, readersMigration); err != nil {
db.Close()
return nil, fmt.Errorf("migrate schema: %w", err)
}
// The owner row must exist before 0004 attaches the existing bookmarks to
// it. The hash refresh is a separate statement after all migrations: the
// token_epoch column 0006 adds does not exist yet at this point, and the
// refresh only ever concerns rows that have never been rotated.
if err := seedOwner(db, owner); err != nil {
db.Close()
return nil, fmt.Errorf("seed owner: %w", err)
}
if err := migrate(db, allMigrations); err != nil {
db.Close()
return nil, fmt.Errorf("migrate: %w", err)
}
if err := refreshOwnerToken(db, owner); err != nil {
db.Close()
return nil, fmt.Errorf("refresh owner token: %w", err)
}
var ownerID int64
if err := db.QueryRow(
`SELECT id FROM readers WHERE discord_id = $1`, owner.DiscordID).Scan(&ownerID); err != nil {
db.Close()
return nil, fmt.Errorf("resolve owner: %w", err)
}
return &Store{
db: db, ownerID: ownerID, coverDir: coverDir, coverBaseURL: base,
}, nil
}
// seedOwner makes sure the configured owner exists as exactly one readers row.
// The hash is only ever written here for a brand-new row; existing rows keep
// what they have until refreshOwnerToken decides otherwise, so the seed can
// never clobber a rotation.
func seedOwner(db *sql.DB, o Owner) error {
if _, err := db.Exec(`
INSERT INTO readers (discord_id, token_sha256) VALUES ($1, $2)
ON CONFLICT (discord_id) DO NOTHING`,
o.DiscordID, o.TokenHash[:]); err != nil {
return fmt.Errorf("seed owner: %w", err)
}
return nil
}
// refreshOwnerToken brings a never-rotated owner row's hash current with the
// configured credential. That is the cutover path: a database seeded under
// the retired global token still carries its hash at epoch 0, and the
// epoch-0 derivation is the caller's TokenHash. A rotated row (epoch > 0) is
// left alone — a restart must not resurrect the old credential by
// overwriting the hash a rotation wrote.
func refreshOwnerToken(db *sql.DB, o Owner) error {
if _, err := db.Exec(`
UPDATE readers SET token_sha256 = $2
WHERE discord_id = $1 AND token_epoch = 0`,
o.DiscordID, o.TokenHash[:]); err != nil {
return fmt.Errorf("refresh owner token: %w", err)
}
return nil
}
// migrate applies every embedded migration this database has not recorded, in
// filename order, each in its own transaction. upto caps the highest version
// applied; 0 means all. Files are named "<version>_<name>.sql" and are
// append-only: editing an applied file changes nothing, because
// schema_migrations is how a database remembers what it ran. Runs on every
// start and is a no-op once current.
func migrate(db *sql.DB, upto int64) error {
if _, err := db.Exec(`CREATE TABLE IF NOT EXISTS schema_migrations (
version bigint PRIMARY KEY,
applied_at timestamptz NOT NULL DEFAULT now())`); err != nil {
return fmt.Errorf("create version table: %w", err)
}
names, err := fs.Glob(migrations, "migrations/*.sql")
if err != nil {
return err
}
slices.Sort(names)
for _, name := range names {
version, err := strconv.ParseInt(strings.SplitN(path.Base(name), "_", 2)[0], 10, 64)
if err != nil {
return fmt.Errorf("migration %q: filename must start with a version number", name)
}
if upto > 0 && version > upto {
continue
}
body, err := migrations.ReadFile(name)
if err != nil {
return err
}
if err := applyMigration(db, version, string(body)); err != nil {
return fmt.Errorf("migration %q: %w", name, err)
}
}
return nil
}
// applyMigration runs one migration and records its version in the same
// transaction, so an interrupted start leaves neither half behind.
func applyMigration(db *sql.DB, version int64, body string) error {
tx, err := db.Begin()
if err != nil {
return err
}
defer tx.Rollback()
var applied bool
if err := tx.QueryRow(
`SELECT EXISTS (SELECT 1 FROM schema_migrations WHERE version = $1)`,
version).Scan(&applied); err != nil {
return err
}
if applied {
return nil
}
// No parameters, so this goes over the simple protocol and a migration may
// hold more than one statement.
if _, err := tx.Exec(body); err != nil {
return err
}
if _, err := tx.Exec(`INSERT INTO schema_migrations (version) VALUES ($1)`, version); err != nil {
return err
}
return tx.Commit()
}
// scanBookmark reads one row in bookmarkColumns order. Every column is NOT
// NULL except latest_chapter_num, where NULL means "never captured" — a
// distinct state from chapter zero, and the reason for the pointer.
func (s *Store) scanBookmark(scan func(...any) error) (Bookmark, error) {
var (
b Bookmark
coverAddress string
latestChapterNum sql.NullFloat64
)
if err := scan(
&b.Site, &b.SeriesID, &b.Title, &b.SeriesURL, &coverAddress,
&b.LastChapter, &b.LastChapterNum, &b.LastChapterURL,
&b.Favorite, &b.LatestChapter, &latestChapterNum, &b.UpdatedAt, &b.Status, &b.Kind,
); err != nil {
return Bookmark{}, err
}
b.Cover = s.CoverWireURL(coverAddress)
if latestChapterNum.Valid {
b.LatestChapterNum = &latestChapterNum.Float64
}
// The wire identity is derived: there is no stored key column, the
// bookmark is keyed (reader_id, site, series_id) (issue #22).
b.Key = b.Site + ":" + b.SeriesID
// An unrecognised bucket (a hand-edited row) would leave the row in no list
// at all, so anything outside the three known buckets reads as the default
// rather than being passed through.
if b.Status != StatusReading && b.Status != StatusArchived && b.Status != StatusFinished {
b.Status = StatusReading
}
return b, nil
}
// scanSeries reads one row in seriesColumns order, plus the due query's
// reader_count column. latest_chapter_num and latest_raised_by are both
// nullable, same as latest_chapter_num on the bookmark read path.
func scanSeries(scan func(...any) error) (Series, error) {
var (
sr Series
latestChapterNum sql.NullFloat64
latestRaisedBy sql.NullInt64
)
if err := scan(
&sr.Site, &sr.SeriesID, &sr.Title, &sr.SeriesURL, &sr.Cover, &sr.CoverAddress,
&sr.Kind, &sr.LatestChapter, &latestChapterNum, &sr.LatestCheckedAt, &latestRaisedBy,
&sr.readerCount,
); err != nil {
return Series{}, err
}
if latestChapterNum.Valid {
sr.LatestChapterNum = &latestChapterNum.Float64
}
if latestRaisedBy.Valid {
sr.LatestRaisedBy = &latestRaisedBy.Int64
}
return sr, nil
}
// Close releases the underlying database handle.
func (s *Store) Close() error { return s.db.Close() }
func coverSourceAddress(sourceURL string) string {
sum := sha256.Sum256([]byte(sourceURL))
return hex.EncodeToString(sum[:])
}
func coverRelativePath(address string) string {
return address[:2] + "/" + address[2:4] + "/" + address
}
func (s *Store) getCover(sourceURL string) ([]byte, string, bool, error) {
return s.getCoverByAddress(coverSourceAddress(sourceURL))
}
func (s *Store) getCoverByAddress(address string) ([]byte, string, bool, error) {
var relativePath, contentType string
err := s.db.QueryRow(
`SELECT path, content_type FROM covers WHERE address = $1`, address,
).Scan(&relativePath, &contentType)
if errors.Is(err, sql.ErrNoRows) {
return nil, "", false, nil
}
if err != nil {
return nil, "", false, fmt.Errorf("get cover %q: %w", address, err)
}
expectedPath := coverRelativePath(address)
if relativePath != expectedPath {
return nil, "", false, fmt.Errorf("cover %q has unexpected path %q", address, relativePath)
}
body, err := os.ReadFile(filepath.Join(s.coverDir, filepath.FromSlash(relativePath)))
if errors.Is(err, fs.ErrNotExist) {
return nil, "", false, nil
}
if err != nil {
return nil, "", false, fmt.Errorf("read cover %q: %w", address, err)
}
return body, contentType, true, nil
}
func (s *Store) putCover(sourceURL string, body []byte, contentType string) error {
stored, ok := CoverContentType(contentType)
if !ok {
return fmt.Errorf("put cover %q: unsupported content type %q", sourceURL, contentType)
}
contentType = stored
address := coverSourceAddress(sourceURL)
relativePath := coverRelativePath(address)
coverPath := filepath.Join(s.coverDir, filepath.FromSlash(relativePath))
if err := os.MkdirAll(filepath.Dir(coverPath), 0o755); err != nil {
return fmt.Errorf("create cover shard: %w", err)
}
tmp, err := os.CreateTemp(filepath.Dir(coverPath), ".cover-*")
if err != nil {
return fmt.Errorf("create cover temp file: %w", err)
}
tmpName := tmp.Name()
defer os.Remove(tmpName)
if _, err := tmp.Write(body); err != nil {
tmp.Close()
return fmt.Errorf("write cover temp file: %w", err)
}
if err := tmp.Sync(); err != nil {
tmp.Close()
return fmt.Errorf("sync cover temp file: %w", err)
}
if err := tmp.Close(); err != nil {
return fmt.Errorf("close cover temp file: %w", err)
}
if err := os.Link(tmpName, coverPath); err != nil && !errors.Is(err, fs.ErrExist) {
return fmt.Errorf("install cover file: %w", err)
}
if _, err := s.db.Exec(`
INSERT INTO covers (address, path, content_type)
VALUES ($1, $2, $3)
ON CONFLICT (address) DO NOTHING`, address, relativePath, contentType); err != nil {
return fmt.Errorf("record cover %q: %w", address, err)
}
return nil
}
// GetCover returns the immutable object addressed by its source URL. Missing
// files are reported with ok=false so callers can retry acquisition later.
func (s *Store) GetCover(sourceURL string) ([]byte, string, bool, error) {
return s.getCover(sourceURL)
}
// PutCover persists bytes under the source URL's content address. A later
// write for the same URL cannot replace the immutable object.
func (s *Store) PutCover(sourceURL string, body []byte, contentType string) error {
return s.putCover(sourceURL, body, contentType)
}
// CoverAddress is the content address bytes fetched from sourceURL are stored
// under. It is a pure function of the URL, so the acquisition path can name a
// Cover before it has the bytes.
func CoverAddress(sourceURL string) string { return coverSourceAddress(sourceURL) }
// coverAddressRe is the shape of a stored address: the hex SHA-256 of a source
// URL. Request paths reach CoverByAddress, so the shape is checked before the
// value is ever turned into a filesystem path.
var coverAddressRe = regexp.MustCompile(`^[0-9a-f]{64}$`)
// CoverByAddress returns the immutable object at one content address. An
// address that is not a stored one - malformed, unknown, or recorded but with
// its file gone - is reported with ok=false rather than as an error.
func (s *Store) CoverByAddress(address string) ([]byte, string, bool, error) {
if !coverAddressRe.MatchString(address) {
return nil, "", false, nil
}
return s.getCoverByAddress(address)
}
// CoverWireURL is the absolute URL a client renders for a stored Cover, and ""
// for a Series that has none yet. A blank is a real state, not a placeholder
// address: it is what tells both clients to draw their own fallback instead of
// requesting bytes that do not exist (ADR-0007).
func (s *Store) CoverWireURL(address string) string {
if address == "" {
return ""
}
return s.coverBaseURL + "/covers/" + address
}
// SetSeriesCover stores the bytes and points the Series at them, but only
// while the Series has no Cover: acquisition at creation and the poll both
// call this, and whichever arrives second must not overwrite the first. The
// bytes themselves are content-addressed and immutable, so storing them twice
// is free.
func (s *Store) SetSeriesCover(site, seriesID, sourceURL string, body []byte, contentType string) error {
if err := s.putCover(sourceURL, body, contentType); err != nil {
return err
}
if _, err := s.db.Exec(`
UPDATE series SET cover = $3, cover_address = $4
WHERE site = $1 AND series_id = $2 AND cover_address = ''`,
site, seriesID, sourceURL, coverSourceAddress(sourceURL)); err != nil {
return fmt.Errorf("set cover for %q: %w", site+":"+seriesID, err)
}
return nil
}
// List returns every bookmark of one reader, newest activity first.
// Series-owned fields are joined in, so each Bookmark reads back whole and
// flat (ADR-0004).
func (s *Store) List(readerID int64) ([]Bookmark, error) {
rows, err := s.db.Query(`SELECT `+bookmarkColumns+`
FROM bookmarks b
JOIN series s ON s.site = b.site AND s.series_id = b.series_id
WHERE b.reader_id = $1
ORDER BY b.updated_at DESC`, readerID)
if err != nil {
return nil, fmt.Errorf("query bookmarks: %w", err)
}
defer rows.Close()
out := []Bookmark{}
for rows.Next() {
b, err := s.scanBookmark(rows.Scan)
if err != nil {
return nil, fmt.Errorf("scan bookmark: %w", err)
}
out = append(out, b)
}
return out, rows.Err()
}
// Get returns one bookmark of one reader by key. A missing key is not an
// error: ok is false and err is nil. UI mutations read-modify-write through
// this so they preserve the fields they do not touch.
func (s *Store) Get(readerID int64, key string) (Bookmark, bool, error) {
site, seriesID, ok := strings.Cut(key, ":")
if !ok {
return Bookmark{}, false, nil
}
b, err := s.scanBookmark(s.db.QueryRow(
`SELECT `+bookmarkColumns+` FROM bookmarks b
JOIN series s ON s.site = b.site AND s.series_id = b.series_id
WHERE b.reader_id = $1 AND b.site = $2 AND b.series_id = $3`,
readerID, site, seriesID).Scan)
if errors.Is(err, sql.ErrNoRows) {
return Bookmark{}, false, nil
}
if err != nil {
return Bookmark{}, false, fmt.Errorf("get %q: %w", key, err)
}
return b, true, nil
}
// Upsert inserts or replaces one reader's bookmark by key (last-write-wins)
// and returns the row as actually stored — one flat object with the
// series-owned fields joined in, exactly as GET reports it (ADR-0004). A
// bookmark is keyed (reader_id, site, series_id), so the same key upserts two
// independent rows for two readers.
//
// The flat body is decomposed across two tables in one transaction. The series
// row is written first (the bookmarks FK requires it to exist), then the
// bookmark row. On the series side, title/series_url/cover are applied only
// when the row is brand new: once a series exists, client-supplied values are
// ignored, because the row is shared and the values are scraped page content —
// see ADR-0003. Kind and the latest-chapter fields are last-write-wins.
//
// b.UpdatedAt is only a candidate: it is applied when the row is new or when
// last_chapter_num changes, and otherwise the stored value is kept. Clients
// order their list by updated_at, so favoriting a series or recording a newly
// published chapter must not disturb that order — only real reading progress
// does. Callers must therefore use the returned bookmark, not the argument.
func (s *Store) Upsert(readerID int64, b Bookmark) (Bookmark, error) {
tx, err := s.db.Begin()
if err != nil {
return Bookmark{}, fmt.Errorf("begin %q: %w", b.Key, err)
}
defer tx.Rollback()
var latestNum any
if b.LatestChapterNum != nil {
latestNum = *b.LatestChapterNum
}
// The kind column resolves on the VALUES side, not in the conflict clause:
// excluded.* is the row *after* these expressions are evaluated, so a
// default applied there would look identical to a real 'manga' and would
// overwrite a novel series on every PUT from a client that knows nothing
// about the column. Resolved once here, an empty incoming kind means "keep
// what is stored", and only a brand-new row falls through to the literal
// default. The subquery runs inside this transaction, so it sees the row
// this statement is about to conflict with. Same pattern as the status
// COALESCE on the bookmark insert below.
//
// The ::text casts are load-bearing: inside COALESCE/NULLIF there is no
// target column to infer the parameter type from, and Postgres rejects the
// statement rather than guessing.
//
// The cover columns are absent on purpose: the Cover is acquired
// server-side (ADR-0007), so a client-supplied one is not written even
// when the row is brand new.
//
// xmax is zero only on a row this statement inserted, which is how a
// Series nobody had bookmarked before is told apart from one that already
// existed — DO UPDATE returns a row either way.
var created bool
if err := tx.QueryRow(`
INSERT INTO series (site, series_id, title, series_url, kind,
latest_chapter, latest_chapter_num)
VALUES ($1, $2, $3, $4,
COALESCE(NULLIF($5::text, ''), (SELECT kind FROM series WHERE site = $1 AND series_id = $2), 'manga'),
$6, $7)
ON CONFLICT (site, series_id) DO UPDATE SET
kind=excluded.kind,
latest_chapter=excluded.latest_chapter,
latest_chapter_num=excluded.latest_chapter_num
RETURNING xmax = 0`,
b.Site, b.SeriesID, b.Title, b.SeriesURL, b.Kind,
b.LatestChapter, latestNum).Scan(&created); err != nil {
return Bookmark{}, fmt.Errorf("upsert series for %q: %w", b.Key, err)
}
// IS DISTINCT FROM is Postgres's null-safe comparison, and it is what
// implements the ordering rule. Within DO UPDATE, a bare column is the
// stored row and excluded.* is the incoming one; a brand-new key never
// reaches this clause, so it keeps the fresh timestamp from VALUES.
if _, err := tx.Exec(`
INSERT INTO bookmarks (reader_id, site, series_id, last_chapter, last_chapter_num,
last_chapter_url, favorite, status, updated_at)
VALUES ($1, $2, $3, $4, $5, $6, $7,
COALESCE(NULLIF($8::text, ''), (SELECT status FROM bookmarks WHERE reader_id = $1 AND site = $2 AND series_id = $3), 'reading'),
$9)
ON CONFLICT (reader_id, site, series_id) DO UPDATE SET
last_chapter=excluded.last_chapter, last_chapter_num=excluded.last_chapter_num,
last_chapter_url=excluded.last_chapter_url,
favorite=excluded.favorite,
status=excluded.status,
updated_at=CASE
WHEN bookmarks.last_chapter_num IS DISTINCT FROM excluded.last_chapter_num
THEN excluded.updated_at
ELSE bookmarks.updated_at
END`,
readerID, b.Site, b.SeriesID,
b.LastChapter, b.LastChapterNum, b.LastChapterURL,
b.Favorite, b.Status, b.UpdatedAt); err != nil {
return Bookmark{}, fmt.Errorf("upsert %q: %w", b.Key, err)
}
stored, err := s.scanBookmark(tx.QueryRow(
`SELECT `+bookmarkColumns+` FROM bookmarks b
JOIN series s ON s.site = b.site AND s.series_id = b.series_id
WHERE b.reader_id = $1 AND b.site = $2 AND b.series_id = $3`,
readerID, b.Site, b.SeriesID).Scan)
if err != nil {
return Bookmark{}, fmt.Errorf("read back %q: %w", b.Key, err)
}
if err := tx.Commit(); err != nil {
return Bookmark{}, fmt.Errorf("commit %q: %w", b.Key, err)
}
// After commit, never inside the transaction: the hook reaches a
// third-party Site, and the Reader's write must not wait on it.
if created && s.OnSeriesCreated != nil {
s.OnSeriesCreated(Series{
Site: b.Site, SeriesID: b.SeriesID, Title: stored.Title,
SeriesURL: stored.SeriesURL, Kind: stored.Kind,
})
}
return stored, nil
}
// Delete removes one reader's bookmark by key. Deleting a missing key is not
// an error.
func (s *Store) Delete(readerID int64, key string) error {
site, seriesID, ok := strings.Cut(key, ":")
if !ok {
return nil
}
if _, err := s.db.Exec(
`DELETE FROM bookmarks WHERE reader_id = $1 AND site = $2 AND series_id = $3`,
readerID, site, seriesID); err != nil {
return fmt.Errorf("delete %q: %w", key, err)
}
return nil
}
// DueForLatestCheck returns one Site's series whose server-side
// latest-chapter check has aged past cutoffMs, ordered by how many bookmarks
// reference them (descending) then least-recently-checked first. One Site per
// query, because each Poll Lane asks for its own list: the query carries one
// Site and one cut-off instead of parallel lists (issue #100). There is no
// limit — the Lane's own gap paces the fetches, and the batch size that used
// to cap this query is gone with the shared pace.
//
// The reader_count ordering is the point of the split (ADR-0003): a series
// shared by several readers is fetched once per due cycle, and the popular
// ones stay freshest while the long tail absorbs any shortfall. Within one
// reader count, oldest-first keeps the poll fair when the backlog outgrows
// throughput: the most neglected series is always next, so a large collection
// refreshes uniformly slower rather than leaving a tail that never refreshes
// at all. The userscript sorts its own queue the same way (L453).
//
// Series with no series_url are skipped — there is nothing to fetch, which is
// the same filter the userscript applies at L452. Series whose only bookmarks
// are finished are skipped too: nothing more is coming, so fetching them only
// burns requests. Archived bookmarks still count — knowing what a shelved
// series is up to is the whole reason for archiving instead of deleting.
// A series with no bookmarks at all never appears: the join excludes it.
//
// ceilingMs is the Sighting deferral ceiling (issue #103): a Series whose last
// real Poll is older than it appears however recently it was sighted. That is
// what bounds the whole mechanism — a wrong Latest Chapter dies within the
// ceiling deterministically rather than in expectation. Deferral itself is
// decided here, from two facts the query already computes, so a Lane gains no
// query per round: a Sighting younger than cutoffMs holds the Series back, but
// only while COUNT(*) is 1. A Series a second Reader bookmarks is Polled on
// schedule, so a wrong value the whole guild can see is corrected by a check
// that was never postponed; on a solitary Series the only person a wrong value
// reaches is the Reader who reported it. Whether the reporting Reader is
// allowed to defer at all was settled when the Sighting was recorded — see
// RecordSighting.
func (s *Store) DueForLatestCheck(site string, cutoffMs, ceilingMs int64) ([]Series, error) {
rows, err := s.db.Query(`SELECT `+seriesColumns+`, COUNT(*) AS reader_count
FROM series s
JOIN bookmarks b ON b.site = s.site AND b.series_id = s.series_id
WHERE s.site = $1
AND s.series_url <> ''
AND s.latest_checked_at <= $2::bigint
GROUP BY s.site, s.series_id, s.title, s.series_url, s.cover,
s.kind, s.latest_chapter, s.latest_chapter_num, s.latest_checked_at
HAVING COUNT(*) FILTER (WHERE b.status <> 'finished') > 0
AND (COUNT(*) > 1
OR s.latest_sighted_at <= $2::bigint
OR s.latest_checked_at <= $3::bigint)
ORDER BY reader_count DESC, s.latest_checked_at ASC`, site, cutoffMs, ceilingMs)
if err != nil {
return nil, fmt.Errorf("query due series: %w", err)
}
defer rows.Close()
out := []Series{}
for rows.Next() {
sr, err := scanSeries(rows.Scan)
if err != nil {
return nil, fmt.Errorf("scan due series: %w", err)
}
out = append(out, sr)
}
return out, rows.Err()
}
// EligibleSeriesCount returns how many of a Site's Series still have at least
// one bookmark outside the finished bucket. It is the denominator of the
// Lane's pace (issue #100): the effective gap is the smaller of the registry
// gap and one hour divided by this count, so Series that will never be Polled
// do not make the Lane faster than it needs to be, and counting every eligible
// Series rather than only those currently due keeps the pace steady — the
// single worst moment to be fastest is startup, when everything is due at
// once.
func (s *Store) EligibleSeriesCount(site string) (int, error) {
var n int
err := s.db.QueryRow(`SELECT COUNT(*) FROM (
SELECT 1
FROM series s
JOIN bookmarks b ON b.site = s.site AND b.series_id = s.series_id
WHERE s.site = $1
GROUP BY s.site, s.series_id
HAVING COUNT(*) FILTER (WHERE b.status <> 'finished') > 0
) e`, site).Scan(&n)
if err != nil {
return 0, fmt.Errorf("count eligible series %s: %w", site, err)
}
return n, nil
}
// MarkLatestChecked records that the server looked at a series at ts, whatever
// the look turned up. Marking a missing series is not an error: the row may
// have been orphaned while a fetch was in flight.
//
// This is the one write that does not go through Upsert, and the column is kept
// out of the client-visible read path on purpose. PUT /bookmarks/{key} decodes
// a whole Bookmark from the client and Upsert writes every series column it
// knows about, so a userscript PUT — which has no idea this field exists —
// would write a zero and reset the rest, making the poller re-fetch that
// series every tick for as long as the user kept reading it.
func (s *Store) MarkLatestChecked(site, seriesID string, ts int64) error {
if _, err := s.db.Exec(
`UPDATE series SET latest_checked_at = $1 WHERE site = $2 AND series_id = $3`,
ts, site, seriesID); err != nil {
return fmt.Errorf("mark checked %s:%s: %w", site, seriesID, err)
}
return nil
}
// LatestCheckedAt reads the column MarkLatestChecked writes. It exists for
// tests outside this package (the poller's own tests assert on rest
// bookkeeping) — see MarkLatestChecked for why the field stays off the
// client-visible row.
func (s *Store) LatestCheckedAt(site, seriesID string) (int64, error) {
var ts int64
if err := s.db.QueryRow(
`SELECT latest_checked_at FROM series WHERE site = $1 AND series_id = $2`,
site, seriesID).Scan(&ts); err != nil {
return 0, fmt.Errorf("latest checked at %s:%s: %w", site, seriesID, err)
}
return ts, nil
}
// SetLatestChapter records the newest chapter the poll found on a series page.
// The poller walks Series rather than Bookmarks, so this is a series-level
// write: the row is shared, and updating it once refreshes every bookmark that
// joins to it. Touching a missing series is not an error.
func (s *Store) SetLatestChapter(site, seriesID, label string, num float64) error {
if _, err := s.db.Exec(
`UPDATE series SET latest_chapter = $3, latest_chapter_num = $4
WHERE site = $1 AND series_id = $2`,
site, seriesID, label, num); err != nil {
return fmt.Errorf("set latest chapter %s:%s: %w", site, seriesID, err)
}
return nil
}
// RecordSighting notes that a Reader's browser reported this Series' Latest
// Chapter, which is the half of a Sighting the client body cannot express
// (issue #103). It must be called *before* the Upsert that stores the reported
// value: the raise test compares against what is still on the row, and after
// the Upsert there is nothing left to compare with. A Series that does not
// exist yet — the first Bookmark of it — is not a Sighting at all: nothing has
// ever been Polled, so there is nothing to defer and nobody to attribute.
//
// Two independent effects, hence the two CASE arms. The deferral stamp is only
// written for a Reader below the disagreement limit, so a marked Reader's
// reports keep updating the Latest Chapter but stop postponing anything, and
// clearing their marks restores the privilege on their next Sighting. The
// attribution is written whenever the report raises the stored number,
// including for a marked Reader — their Sightings are still judged, which is
// how they earn the privilege back.
//
// num is the reported chapter number. A PUT that carries none — a favourite
// toggle, or progress written from a chapter page — is no Sighting at all:
// nobody read the Series page, so there is nothing to stand in for a Poll and
// nothing that could later be judged.
func (s *Store) RecordSighting(readerID int64, site, seriesID string, num *float64, ts int64) error {
if num == nil {
return nil
}
if _, err := s.db.Exec(`
UPDATE series SET
latest_sighted_at = CASE
WHEN (SELECT sighting_disagreements FROM readers WHERE id = $3) < $6
THEN $4::bigint ELSE latest_sighted_at END,
latest_raised_by = CASE
WHEN latest_chapter_num IS NULL OR $5::double precision > latest_chapter_num
THEN $3::bigint ELSE latest_raised_by END
WHERE site = $1 AND series_id = $2`,
site, seriesID, readerID, ts, *num, SightingDisagreementLimit); err != nil {
return fmt.Errorf("record sighting %s:%s: %w", site, seriesID, err)
}
return nil
}
// SightingAgreementsToClear is how many Polls must confirm a Reader's
// Sightings in a row before their disagreements are forgiven. An agreement is
// only recorded when a Poll later confirms a Sighting, so this is twenty Polls
// of Series that Reader bookmarks — hours to days, not twenty page views. That
// is the intended price: recovery is automatic but cannot be outwaited, and a
// disagreement resets the run to zero, so credit cannot be banked in advance.
const SightingAgreementsToClear = 20
// RecordSightingOutcome settles what a Poll decided about the Reader whose
// Sighting last raised this Series' Latest Chapter, and clears the attribution
// in the same transaction so one Sighting is judged exactly once. agreed is
// the Poll confirming the stored value; its opposite is the Poll finding a
// lower number, which means the raise was false.
//
// A Poll finding a *higher* number is neither — the Site published — and takes
// ClearSightingAttribution instead.
func (s *Store) RecordSightingOutcome(site, seriesID string, readerID int64, agreed bool) error {
tx, err := s.db.Begin()
if err != nil {
return fmt.Errorf("begin sighting outcome %s:%s: %w", site, seriesID, err)
}
defer tx.Rollback()
// The run length is what "consecutive" means: a disagreement zeroes the
// agreements, and completing a run zeroes both, so the next run starts
// from nothing rather than forgiving every later disagreement instantly.
q := `UPDATE readers SET sighting_disagreements = sighting_disagreements + 1,
sighting_agreements = 0
WHERE id = $1`
args := []any{readerID}
if agreed {
q = `UPDATE readers SET
sighting_agreements = CASE WHEN sighting_agreements + 1 >= $2 THEN 0
ELSE sighting_agreements + 1 END,
sighting_disagreements = CASE WHEN sighting_agreements + 1 >= $2 THEN 0
ELSE sighting_disagreements END
WHERE id = $1`
args = append(args, SightingAgreementsToClear)
}
if _, err := tx.Exec(q, args...); err != nil {
return fmt.Errorf("record sighting outcome for reader %d: %w", readerID, err)
}
if _, err := tx.Exec(clearAttributionSQL, site, seriesID, readerID); err != nil {
return fmt.Errorf("clear sighting attribution %s:%s: %w", site, seriesID, err)
}
if err := tx.Commit(); err != nil {
return fmt.Errorf("commit sighting outcome %s:%s: %w", site, seriesID, err)
}
return nil
}
// ClearSightingAttribution answers a Sighting without judging it: the Poll
// found a higher number, so the value about to be stored is its own and this
// Reader is no longer answerable for the row. Without it the next Poll's
// agreement would be credited to a Reader who did not earn it.
func (s *Store) ClearSightingAttribution(site, seriesID string, readerID int64) error {
if _, err := s.db.Exec(clearAttributionSQL, site, seriesID, readerID); err != nil {
return fmt.Errorf("clear sighting attribution %s:%s: %w", site, seriesID, err)
}
return nil
}
// clearAttributionSQL drops the attribution only while it still names the
// Reader being judged: a Sighting landing between the due query's snapshot and
// this write is a fresh, unjudged one and must not be erased by the previous
// one's verdict.
const clearAttributionSQL = `UPDATE series SET latest_raised_by = NULL
WHERE site = $1 AND series_id = $2 AND latest_raised_by = $3`