mangaBookmark/backend/internal/latest/poller.go

package latest

import (
	"context"
	"errors"
	"log"
	"net/url"
	"time"

	"bookmarkmanager/backend/internal/store"
)

// Fetcher retrieves a series page. It exists as an interface so tests can inject
// a fake: nothing in the test suite may touch the network or the TLS client.
type Fetcher interface {
	Get(ctx context.Context, url string) (body string, status int, err error)
}

// BrowserCoverFetcher retrieves one cover's bytes through the browser-backed
// path — the only route that clears the challenge kagane's image URLs answer
// a plain fetch with. Satisfied by BrowserFetcher.
type BrowserCoverFetcher interface {
	Image(ctx context.Context, imageURL string) (body []byte, contentType string, err error)
}

// Poller re-checks each bookmarked series' newest published chapter on a
// schedule, independent of the userscript's own in-browser checks. The two run
// in parallel and report the same observable fact, so whichever writes last wins
// and neither needs to know about the other.
//
// Two clocks, deliberately independent:
//
//   - Interval is how often this goroutine wakes up and looks.
//   - Cooldowns are how long a series rests since its own last check. Browser-
//     backed sites use the longer BrowserCooldown.
//
// Cooldowns are enforced by the WHERE clause in DueForLatestCheck rather than
// by any timer. Shortening Interval therefore cannot shorten anyone's cooldown;
// it only makes the poller wake up and find nothing due more often.
type Poller struct {
	Store *store.Store
	Fetch Fetcher
	// BrowserFetch handles sites behind a JavaScript challenge that Fetch
	// cannot clear. Nil disables those sites entirely rather than falling back
	// to Fetch, which would only ever retrieve a challenge page.
	BrowserFetch Fetcher
	// CoverFetch is optional; failures are logged and never affect the chapter poll.
	CoverFetch BrowserCoverFetcher
	// CoverBytesFetch is optional; it handles plain-TLS sources through the
	// same failure-isolated prefetch path.
	CoverBytesFetch CoverBytesFetcher
	Now             func() time.Time // injected so tests can freeze it
	Cooldown        time.Duration
	BrowserCooldown time.Duration
	Interval        time.Duration
	Stagger         time.Duration
	Batch           int
}

// fillBlankCover gives a Series its Cover when it has none. The blank state is
// what "no Cover yet" means on the wire (ADR-0007): permanently-blank rows
// created before acquisition existed, and rows whose creation-time fetch
// failed, both heal here. A non-blank CoverAddress is left alone — refetching
// would add a request per Series per cycle and change artwork under the Reader
// for no visible reason. A row that already carries a source URL is owned by
// prefetchCover instead; this path only records a Cover address already
// extracted from the series page.
//
// Failures are logged against the Series and never returned: the chapter poll
// must not notice. A failed fill is retried the next time this Series is due;
// there is no separate retry queue.
func (p *Poller) fillBlankCover(ctx context.Context, sr store.Series, cover string) {
	if sr.CoverAddress != "" || sr.Cover != "" {
		return
	}
	if cover == "" {
		return
	}
	p.storeCover(ctx, sr, cover)
}

// prefetchCover heals Series that already carry a third-party source URL but
// no stored address — the state left by client-supplied covers before
// acquisition moved server-side. Every Site takes the same path; fetchCoverBytes
// routes by URL shape, so browser-claimed URLs still need the sidecar. New
// blanks have no source URL and go through fillBlankCover from the series page
// instead.
func (p *Poller) prefetchCover(ctx context.Context, sr store.Series) {
	if sr.Cover == "" || sr.CoverAddress != "" {
		return
	}
	body, contentType, found, err := p.Store.GetCover(sr.Cover)
	if err != nil {
		log.Printf("latest poll %q: read cover: %v", sr.Key(), err)
		return
	}
	if found {
		if err := p.Store.SetSeriesCover(sr.Site, sr.SeriesID, sr.Cover, body, contentType); err != nil {
			log.Printf("latest poll %q: persist cover: %v", sr.Key(), err)
		}
		return
	}
	p.storeCover(ctx, sr, sr.Cover)
}

// storeCover fetches bytes for sourceURL and points the Series at them. Every
// failure is logged against the Series and swallowed so the chapter poll
// cannot see it.
func (p *Poller) storeCover(ctx context.Context, sr store.Series, sourceURL string) {
	bytes, contentType, err := fetchCoverBytes(ctx, sourceURL, p.CoverFetch, p.CoverBytesFetch)
	if err != nil {
		log.Printf("latest poll %q: fetch cover %s: %v", sr.Key(), sourceURL, err)
		return
	}
	if err := p.Store.SetSeriesCover(sr.Site, sr.SeriesID, sourceURL, bytes, contentType); err != nil {
		log.Printf("latest poll %q: persist cover: %v", sr.Key(), err)
	}
}

// fetcherFor returns the fetcher a site's page needs, or nil when the site
// cannot be fetched at all right now. A Site whose registry entry carries a
// Browser read — kagane and novelfull, both behind a Cloudflare JavaScript
// challenge no TLS fingerprint clears — prefers the browser; when it is
// absent, the entry's Fallback decides whether plain TLS may take over. One
// routing rule for the poll and the acquirer, so the two cannot drift apart.
func fetcherFor(site string, browser, tls Fetcher) Fetcher {
	s, known := sites[site]
	if !known {
		// No registry entry means nothing to fetch or parse; fail closed even
		// though the only caller gates first, so a future caller that skips
		// the gate cannot hand an arbitrary https URL to the TLS fetcher.
		return nil
	}
	if s.Browser == nil {
		return tls
	}
	if browser != nil {
		return browser
	}
	if s.Browser.Fallback {
		return tls
	}
	return nil
}

// Run polls until ctx is cancelled.
//
// runOnce is called synchronously, so a batch that overruns the tick delays the
// next one instead of stacking a second batch on top of it. That is the intended
// failure mode for a misconfigured batch x stagger: a slower cadence, never
// concurrent fetch storms.
func (p *Poller) Run(ctx context.Context) {
	log.Printf("latest-chapter poller: interval=%s cooldown=%s browser-cooldown=%s batch=%d stagger=%s",
		p.Interval, p.Cooldown, p.BrowserCooldown, p.Batch, p.Stagger)
	t := time.NewTicker(p.Interval)
	defer t.Stop()
	for {
		select {
		case <-ctx.Done():
			log.Println("latest-chapter poller: stopped")
			return
		case <-t.C:
			p.runOnce(ctx)
		}
	}
}

// runOnce processes one batch of due series.
func (p *Poller) runOnce(ctx context.Context) {
	now := p.Now()
	cutoff := now.Add(-p.Cooldown).UnixMilli()
	browserCutoff := now.Add(-p.BrowserCooldown).UnixMilli()
	due, err := p.Store.DueForLatestCheck(cutoff, browserCutoff, browserBackedSites(), p.Batch)
	if err != nil {
		log.Printf("latest poll: due query: %v", err)
		return
	}

	checked := 0
	for i, sr := range due {
		if ctx.Err() != nil {
			break
		}
		// Staggered rather than fired together: a burst of simultaneous requests
		// from one server IP is the traffic shape most likely to move that IP's
		// bot score. This is the server-side analogue of the userscript's "one
		// series per navigation ... indistinguishable from browsing" (L455-456).
		stopped := false
		if i > 0 && p.Stagger > 0 {
			select {
			case <-ctx.Done():
				stopped = true
			case <-time.After(p.Stagger):
			}
		}
		if stopped {
			break
		}
		p.checkOne(ctx, sr)
		checked++
	}
	// due vs checked is how you tell which constraint is binding: ticks that
	// report due=0 mean the cooldown is the limit, ticks that report due==batch
	// every time mean throughput is.
	log.Printf("latest poll: due=%d checked=%d", len(due), checked)
}

// checkOne re-checks one series. Every failure path here is "log and move on":
// the poller is a best-effort enhancement, and no single bad series may stall a
// batch or take down the process.
func (p *Poller) checkOne(ctx context.Context, sr store.Series) {
	defer func() {
		if r := recover(); r != nil {
			log.Printf("latest poll %q: recovered from panic: %v", sr.Key(), r)
		}
	}()

	// Stamped before the fetch, not after, so an error, a timeout, or a shutdown
	// mid-request still consumes the cooldown. Otherwise a renamed or deleted
	// series would be retried on every single tick forever. The userscript
	// stamps in the same order and for the same reason (L471-473).
	if err := p.Store.MarkLatestChecked(sr.Site, sr.SeriesID, p.Now().UnixMilli()); err != nil {
		log.Printf("latest poll %q: mark checked: %v", sr.Key(), err)
		return
	}

	facts, err := readSeriesPage(ctx, sr.Site, sr.SeriesURL, p.BrowserFetch, p.Fetch)
	if err != nil {
		switch {
		case errors.Is(err, errNotFetchable):
			// The cooldown above is already consumed, so a row that never
			// passes the gate is retried at cooldown pace rather than
			// hot-looping.
			log.Printf("latest poll %q: not fetchable: site=%q url=%q", sr.Key(), sr.Site, sr.SeriesURL)
			return
		case errors.Is(err, errNoFetcher):
			log.Printf("latest poll %q: no fetcher for site %q", sr.Key(), sr.Site)
			return
		}
		// A legacy cover heals independently of the page read: its source may
		// answer — a CDN — while the origin does not, so a fetch failure does
		// not skip the heal, matching the order the shared read replaced.
		p.prefetchCover(ctx, sr)
		log.Printf("latest poll %q: %v", sr.Key(), err)
		return
	}
	// A legacy cover source is healed independently of the page read.
	p.prefetchCover(ctx, sr)
	// Cover fill is independent of the chapter signal: a page that lost its
	// chapter list may keep its og:image, and a blank Series heals either way.
	p.fillBlankCover(ctx, sr, facts.Cover)
	if !facts.HasLatest {
		// Most likely a challenge page or a layout change. Either way the row is
		// already stamped, so this waits out a cooldown instead of hot-looping.
		log.Printf("latest poll %q: no chapter links in %d bytes", sr.Key(), facts.BodyLen)
		return
	}

	// Equality, not >, mirroring the userscript (L427): a site that retracts a
	// chapter should correct the stored number downward. The comparison is
	// against the due-query snapshot; a concurrent write in between only costs
	// one redundant UPDATE of the same absolute value, never a wrong one.
	if sr.LatestChapterNum != nil && *sr.LatestChapterNum == facts.Latest.Num {
		return
	}

	// Series-level write: the row is shared, so one update refreshes every
	// bookmark joining to it, and the bookmark's updated_at is never touched —
	// a newly published chapter is not reading progress and must not reorder
	// the list.
	if err := p.Store.SetLatestChapter(sr.Site, sr.SeriesID, facts.Latest.Label, facts.Latest.Num); err != nil {
		log.Printf("latest poll %q: set latest chapter: %v", sr.Key(), err)
		return
	}
	log.Printf("latest poll %q: latest is now %s", sr.Key(), facts.Latest.Label)
}

// fetchableSeriesURL reports whether site is a Site the registry knows and
// seriesURL is safe to hand to a fetcher: an https URL whose host matches the
// Site's pinned hostname exactly. series_url comes from client-supplied PUT
// bodies, so this is a defence against the poller being used to probe
// arbitrary hosts from the server's own network position, not just a check
// against wasted requests. The pin guards different things per Site — a
// browser Site guards a control that executes JavaScript and carries cookies,
// a parser Site guards a wasted request — but the rule is one rule, from the
// registry.
func fetchableSeriesURL(site, seriesURL string) bool {
	s, known := sites[site]
	if !known {
		return false
	}
	u, err := url.Parse(seriesURL)
	if err != nil {
		return false
	}
	return u.Scheme == "https" && u.Hostname() == s.Host
}