package store

import (
	"reflect"
	"strconv"
	"strings"
	"testing"
)

// seriesSeed describes one Series (and optionally its bookmarks) to stand up
// for an admin filter test. Direct SQL, because the filters separate rows the
// Upsert path could not produce together: an orphan has no bookmark, and a
// Reader-raised Latest Chapter needs a Sighting the store does not create.
type seriesSeed struct {
	key             string
	kind            string
	url             string
	cover           string // cover_address
	checkedAt       int64
	latestNum       *float64
	bookmarks       int   // readers that hold it; 0 = orphan
	raisedBy        bool  // a Reader's report is attributed as the raiser
	siteCompletedAt int64 // the Site's own marker (issue #168), 0 = not set
}

// seedAdminSeries inserts one series row and its bookmarks (owner first, then
// fresh readers), with the exact admin-relevant facts a test needs.
func seedAdminSeries(t *testing.T, s *Store, seed seriesSeed) {
	t.Helper()
	site, seriesID, ok := strings.Cut(seed.key, ":")
	if !ok {
		t.Fatalf("key %q: no ':' separator", seed.key)
	}
	if seed.kind == "" {
		seed.kind = "manga"
	}
	var latestChapter any = ""
	if seed.latestNum != nil {
		latestChapter = "Chapter " + strconv.FormatFloat(*seed.latestNum, 'f', -1, 64)
	}
	if _, err := s.db.Exec(`
		INSERT INTO series (site, series_id, title, kind, series_url, cover_address,
		                    latest_checked_at, latest_chapter, latest_chapter_num, site_completed_at)
		VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10)`,
		site, seriesID, "Title of "+seed.key, seed.kind, seed.url, seed.cover,
		seed.checkedAt, latestChapter, seed.latestNum, seed.siteCompletedAt); err != nil {
		t.Fatalf("seed series %q: %v", seed.key, err)
	}
	for i := range seed.bookmarks {
		var readerID int64 = s.OwnerID()
		if i > 0 {
			readerID = secondReader(t, s)
		}
		if _, err := s.db.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, '', 0, '', false, 'reading', $4)`,
			readerID, site, seriesID, seed.checkedAt); err != nil {
			t.Fatalf("seed bookmark %q: %v", seed.key, err)
		}
	}
	if seed.raisedBy {
		if _, err := s.db.Exec(
			`UPDATE series SET latest_raised_by = $1 WHERE site = $2 AND series_id = $3`,
			s.OwnerID(), site, seriesID); err != nil {
			t.Fatalf("seed raised-by %q: %v", seed.key, err)
		}
	}
}

func pageKeys(t *testing.T, s *Store, f SeriesFilter) map[string]bool {
	t.Helper()
	page, err := s.SeriesPage(f)
	if err != nil {
		t.Fatalf("SeriesPage(%+v): %v", f, err)
	}
	keys := map[string]bool{}
	for _, a := range page.Rows {
		keys[a.Key()] = true
	}
	return keys
}

// Each filter must return the rows it names and no others, over one shared
// seeded mix where every healthy neighbour is present to be wrongly returned.
// The stale cutoff is 5000: a Series checked at 9000 is current, at 2000 stale.
func TestAdminSeriesFilters(t *testing.T) {
	s := newTestStore(t)
	seedAdminSeries(t, s, seriesSeed{key: "asura:healthy", url: "https://asurascans.com/comics/healthy", cover: "aaa", checkedAt: 9000, latestNum: new(10.0), bookmarks: 1})
	seedAdminSeries(t, s, seriesSeed{key: "asura:nourl", url: "", cover: "bbb", checkedAt: 9000, latestNum: new(5.0), bookmarks: 1})
	seedAdminSeries(t, s, seriesSeed{key: "asura:nochapter", url: "https://asurascans.com/comics/nochapter", cover: "ccc", checkedAt: 9000, bookmarks: 1})
	seedAdminSeries(t, s, seriesSeed{key: "asura:neverchecked", url: "https://asurascans.com/comics/neverchecked", cover: "ddd", checkedAt: 0, bookmarks: 1})
	seedAdminSeries(t, s, seriesSeed{key: "asura:orphan", url: "https://asurascans.com/comics/orphan", cover: "eee", checkedAt: 9000, latestNum: new(7.0), bookmarks: 0})
	seedAdminSeries(t, s, seriesSeed{key: "asura:stale", url: "https://asurascans.com/comics/stale", cover: "fff", checkedAt: 2000, latestNum: new(4.0), bookmarks: 1})
	seedAdminSeries(t, s, seriesSeed{key: "asura:nocover", url: "https://asurascans.com/comics/nocover", checkedAt: 9000, latestNum: new(9.0), bookmarks: 1})
	seedAdminSeries(t, s, seriesSeed{key: "asura:report", url: "https://asurascans.com/comics/report", cover: "ggg", checkedAt: 9000, latestNum: new(8.0), bookmarks: 1, raisedBy: true})

	cases := []struct {
		name string
		f    SeriesFilter
		want []string
	}{
		{"all", SeriesFilter{}, []string{"asura:healthy", "asura:nourl", "asura:nochapter", "asura:neverchecked", "asura:orphan", "asura:stale", "asura:nocover", "asura:report"}},
		{"no series url", SeriesFilter{Name: SeriesFilterNoURL}, []string{"asura:nourl"}},
		{"never read a chapter", SeriesFilter{Name: SeriesFilterNoChapter}, []string{"asura:nochapter"}},
		{"never checked", SeriesFilter{Name: SeriesFilterNeverChecked}, []string{"asura:neverchecked"}},
		{"no readers", SeriesFilter{Name: SeriesFilterNoReaders}, []string{"asura:orphan"}},
		{"stale", SeriesFilter{Name: SeriesFilterStale, Cutoff: 5000}, []string{"asura:stale"}},
		{"no cover", SeriesFilter{Name: SeriesFilterNoCover}, []string{"asura:nocover"}},
		{"reader report", SeriesFilter{Name: SeriesFilterReaderReport}, []string{"asura:report"}},
	}
	for _, tc := range cases {
		t.Run(tc.name, func(t *testing.T) {
			got := pageKeys(t, s, tc.f)
			want := map[string]bool{}
			for _, k := range tc.want {
				want[k] = true
			}
			if len(got) != len(want) {
				t.Fatalf("%+v returned %v, want exactly %v", tc.f, got, want)
			}
			for k := range want {
				if !got[k] {
					t.Fatalf("%+v dropped %q (got %v)", tc.f, k, got)
				}
			}
		})
	}
}

// A finished Series is the owner's deliberate state, not a problem a Poll
// will fix: the four clock-driven hygiene predicates exclude it (their
// stamps stop advancing at the last Poll, so without the guard a retired row
// is reported forever), the three outcome-driven ones still include it, and
// the finished filter returns exactly the retired rows.
func TestAdminFinishedSeriesFilters(t *testing.T) {
	s := newTestStore(t)
	// Each fin-* row is shaped to trip exactly one predicate if its guard
	// fails: checked-but-old for stale, a zero stamp for never-checked, a
	// stamp with no chapter for no-chapter, an empty cover for no-cover, and
	// the unguarded three shaped to trip their own. A healthy, unfinished
	// neighbour keeps the exclusion checks honest: a filter that regressed to
	// matching nothing would pass a bare "no finished rows" assertion.
	seedAdminSeries(t, s, seriesSeed{key: "asura:fin-stale", url: "u", cover: "c", checkedAt: 2000, latestNum: new(4.0), bookmarks: 1})
	seedAdminSeries(t, s, seriesSeed{key: "asura:fin-neverchecked", url: "u", cover: "c", checkedAt: 0, latestNum: new(4.0), bookmarks: 1})
	seedAdminSeries(t, s, seriesSeed{key: "asura:fin-nochapter", url: "u", cover: "c", checkedAt: 9000, bookmarks: 1})
	seedAdminSeries(t, s, seriesSeed{key: "asura:fin-nocover", url: "u", checkedAt: 9000, latestNum: new(4.0), bookmarks: 1})
	seedAdminSeries(t, s, seriesSeed{key: "asura:fin-nourl", url: "", cover: "c", checkedAt: 9000, latestNum: new(4.0), bookmarks: 1})
	seedAdminSeries(t, s, seriesSeed{key: "asura:fin-orphan", url: "u", cover: "c", checkedAt: 9000, latestNum: new(4.0), bookmarks: 0})
	seedAdminSeries(t, s, seriesSeed{key: "asura:fin-report", url: "u", cover: "c", checkedAt: 9000, latestNum: new(4.0), bookmarks: 1, raisedBy: true})
	seedAdminSeries(t, s, seriesSeed{key: "asura:healthy", url: "u", cover: "c", checkedAt: 9000, latestNum: new(4.0), bookmarks: 1})
	finished := []string{
		"asura:fin-stale", "asura:fin-neverchecked", "asura:fin-nochapter",
		"asura:fin-nocover", "asura:fin-nourl", "asura:fin-orphan", "asura:fin-report",
	}
	for _, key := range finished {
		site, id, _ := strings.Cut(key, ":")
		if err := s.SetSeriesFinished(site, id, 1000); err != nil {
			t.Fatalf("finish %s: %v", key, err)
		}
	}

	cases := []struct {
		name string
		f    SeriesFilter
		want []string
	}{
		{"stale excludes finished", SeriesFilter{Name: SeriesFilterStale, Cutoff: 5000}, nil},
		{"never checked excludes finished", SeriesFilter{Name: SeriesFilterNeverChecked}, nil},
		{"no chapter excludes finished", SeriesFilter{Name: SeriesFilterNoChapter}, nil},
		{"no cover excludes finished", SeriesFilter{Name: SeriesFilterNoCover}, nil},
		{"no url includes finished", SeriesFilter{Name: SeriesFilterNoURL}, []string{"asura:fin-nourl"}},
		{"no readers includes finished", SeriesFilter{Name: SeriesFilterNoReaders}, []string{"asura:fin-orphan"}},
		{"reader report includes finished", SeriesFilter{Name: SeriesFilterReaderReport}, []string{"asura:fin-report"}},
		{"finished returns the retired rows", SeriesFilter{Name: SeriesFilterFinished}, finished},
	}
	for _, tc := range cases {
		t.Run(tc.name, func(t *testing.T) {
			got := pageKeys(t, s, tc.f)
			want := map[string]bool{}
			for _, k := range tc.want {
				want[k] = true
			}
			if len(got) != len(want) {
				t.Fatalf("%+v returned %v, want exactly %v", tc.f, got, want)
			}
			for k := range want {
				if !got[k] {
					t.Fatalf("%+v dropped %q (got %v)", tc.f, k, got)
				}
			}
		})
	}

	// The aggregate's finished total counts every retired row — the same
	// predicate the Overview's finished figure is summed from.
	shapes, err := s.SeriesShapes(SeriesFilter{Name: SeriesFilterFinished})
	if err != nil {
		t.Fatalf("SeriesShapes(finished): %v", err)
	}
	sum := 0
	for _, sh := range shapes {
		sum += sh.Total
	}
	if sum != len(finished) {
		t.Fatalf("finished aggregate = %d, want %d", sum, len(finished))
	}
}

// "Never read a chapter" and "never checked" are disjoint by construction:
// the first requires a non-zero check stamp, the second a zero one. Over a
// mix that should satisfy both, no row may be counted twice.
func TestAdminNeverChapterAndNeverCheckedAreDisjoint(t *testing.T) {
	s := newTestStore(t)
	seedAdminSeries(t, s, seriesSeed{key: "asura:nochapter", url: "u", checkedAt: 9000, bookmarks: 1})
	seedAdminSeries(t, s, seriesSeed{key: "asura:neverchecked", url: "u", checkedAt: 0, bookmarks: 1})
	// A zero-stamp, no-chapter row is never-checked only: if never-read-a-
	// chapter ever lost its non-zero-stamp guard, it would claim this row too
	// and the two counts would double-report it.
	seedAdminSeries(t, s, seriesSeed{key: "asura:both", url: "u", checkedAt: 0, bookmarks: 1})
	noChapter := pageKeys(t, s, SeriesFilter{Name: SeriesFilterNoChapter})
	neverChecked := pageKeys(t, s, SeriesFilter{Name: SeriesFilterNeverChecked})
	for k := range noChapter {
		if neverChecked[k] {
			t.Fatalf("row %q matches both never-read-a-chapter and never-checked", k)
		}
	}
	if !noChapter["asura:nochapter"] || !neverChecked["asura:neverchecked"] {
		t.Fatalf("disjoint split lost its own rows: no-chapter=%v never-checked=%v", noChapter, neverChecked)
	}
}

// Several rows share a zero check stamp, so ordering on latest_checked_at
// alone gives no stable page boundary. The (site, series_id) tie-break must
// make page 2 a strict continuation of page 1: no repeat, no vanishing row.
func TestAdminSeriesPageTieBreakIsStable(t *testing.T) {
	s := newTestStore(t)
	const total = 53 // > one page, < two (page size 50)
	for i := range total {
		id := "tie" + strconv.Itoa(i)
		seedAdminSeries(t, s, seriesSeed{key: "asura:" + id, url: "u", checkedAt: 0, bookmarks: 1})
	}
	// A second Site's zero-stamp row is part of the same all-filter list, and
	// must land on a valid page boundary rather than duplicating or dropping
	// one of asura's rows: the tie-break is global (site, series_id).
	seedAdminSeries(t, s, seriesSeed{key: "demonic:z", url: "u", checkedAt: 0, bookmarks: 1})
	wantTotal := total + 1

	p1, err := s.SeriesPage(SeriesFilter{Name: SeriesFilterAll})
	if err != nil {
		t.Fatalf("SeriesPage page 1: %v", err)
	}
	p2, err := s.SeriesPage(SeriesFilter{Name: SeriesFilterAll, Page: 2})
	if err != nil {
		t.Fatalf("SeriesPage page 2: %v", err)
	}
	if len(p1.Rows) != seriesPageSize {
		t.Fatalf("page 1 has %d rows, want %d", len(p1.Rows), seriesPageSize)
	}
	seen := map[string]bool{}
	for _, a := range append(append([]AdminSeries{}, p1.Rows...), p2.Rows...) {
		if seen[a.Key()] {
			t.Fatalf("row %q repeats across pages", a.Key())
		}
		seen[a.Key()] = true
	}
	if len(seen) != wantTotal {
		t.Fatalf("%d distinct rows across pages, want %d (a row vanished)", len(seen), wantTotal)
	}
	if p1.Total != wantTotal {
		t.Fatalf("page total = %d, want %d (the window count must span pages)", p1.Total, wantTotal)
	}
	// A page beyond the end is empty, not an error (the list re-reads page 1).
	// The window count runs over the rows present in the result set, so an
	// overflow page has no rows and therefore no total — the caller must not
	// render it, which is exactly why the list re-reads page 1.
	pFinal, err := s.SeriesPage(SeriesFilter{Name: SeriesFilterAll, Page: 99})
	if err != nil {
		t.Fatalf("SeriesPage beyond end: %v", err)
	}
	if len(pFinal.Rows) != 0 {
		t.Fatalf("page beyond end = %d rows, want 0", len(pFinal.Rows))
	}
}

// The filtered total is the window number over the same filter the rows use,
// and the per-Site aggregate sums to the same figure — so the landing page's
// count and the list's heading can never disagree, whoever computes them.
func TestAdminTotalAgreesWithRowsAndShapes(t *testing.T) {
	s := newTestStore(t)
	seedAdminSeries(t, s, seriesSeed{key: "asura:a", url: "u", cover: "a", checkedAt: 9000, bookmarks: 1})
	seedAdminSeries(t, s, seriesSeed{key: "asura:b", url: "u", checkedAt: 9000, bookmarks: 1})
	seedAdminSeries(t, s, seriesSeed{key: "asura:c", url: "u", checkedAt: 9000, bookmarks: 1, latestNum: new(2.0), raisedBy: true})
	seedAdminSeries(t, s, seriesSeed{key: "demonic:d", url: "u", kind: "novel", checkedAt: 9000, bookmarks: 1})

	filters := []SeriesFilter{
		{},
		{Name: SeriesFilterNoCover},
		{Name: SeriesFilterReaderReport},
		{Name: SeriesFilterNoChapter},
	}
	for _, f := range filters {
		page, err := s.SeriesPage(f)
		if err != nil {
			t.Fatalf("SeriesPage(%+v): %v", f, err)
		}
		want := len(page.Rows)
		if f.Page == 0 && want == seriesPageSize {
			t.Fatalf("seed produced a full page; bump the seed or drop page size in the test")
		}
		if page.Total != want {
			t.Fatalf("%+v total = %d, want %d (window count disagrees with row count)", f, page.Total, want)
		}
		shapes, err := s.SeriesShapes(f)
		if err != nil {
			t.Fatalf("SeriesShapes(%+v): %v", f, err)
		}
		sum := 0
		for _, sh := range shapes {
			sum += sh.Total
		}
		if sum != want {
			t.Fatalf("%+v aggregate sum = %d, want %d (aggregate disagrees with row query)", f, sum, want)
		}
	}

	// The default filter's aggregate carries the library shape: per-Site
	// totals and the manga/novel split, summed in Go for library wide.
	shapes, err := s.SeriesShapes(SeriesFilter{})
	if err != nil {
		t.Fatalf("SeriesShapes default: %v", err)
	}
	if len(shapes) != 2 || shapes[0].Site != "asura" || shapes[1].Site != "demonic" {
		t.Fatalf("shapes = %+v, want asura then demonic", shapes)
	}
	if shapes[0].Total != 3 || shapes[0].Manga != 3 || shapes[0].Novel != 0 {
		t.Fatalf("asura shape = %+v, want 3 manga, 0 novel", shapes[0])
	}
	if shapes[1].Total != 1 || shapes[1].Manga != 0 || shapes[1].Novel != 1 {
		t.Fatalf("demonic shape = %+v, want 1 novel", shapes[1])
	}
}

// Site and Library narrowing stack on a named filter without changing what
// the filter means.
func TestAdminFilterSiteAndKindNarrow(t *testing.T) {
	s := newTestStore(t)
	seedAdminSeries(t, s, seriesSeed{key: "asura:aa", url: "u", checkedAt: 9000, bookmarks: 1, latestNum: new(1.0)})
	seedAdminSeries(t, s, seriesSeed{key: "asura:ab", url: "", checkedAt: 9000, bookmarks: 1})
	seedAdminSeries(t, s, seriesSeed{key: "demonic:aa", url: "u", kind: "novel", checkedAt: 9000, bookmarks: 1})
	seedAdminSeries(t, s, seriesSeed{key: "demonic:ab", url: "", kind: "novel", checkedAt: 9000, bookmarks: 1})

	got := pageKeys(t, s, SeriesFilter{Name: SeriesFilterNoURL, Site: "asura"})
	if len(got) != 1 || !got["asura:ab"] {
		t.Fatalf("site+nourl = %v, want only asura:ab", got)
	}
	got = pageKeys(t, s, SeriesFilter{Name: SeriesFilterNoURL, Kind: "novel"})
	if len(got) != 1 || !got["demonic:ab"] {
		t.Fatalf("kind+nourl = %v, want only demonic:ab", got)
	}
	got = pageKeys(t, s, SeriesFilter{Name: SeriesFilterAll, Site: "demonic", Kind: "novel"})
	if len(got) != 2 || !got["demonic:aa"] || !got["demonic:ab"] {
		t.Fatalf("site+kind+all = %v, want both demonic rows", got)
	}

	// The aggregate ignores the Site narrowing (it is per-Site by shape), but
	// honours the Library narrowing: asura's missing-URL row is manga, so the
	// novel no-URL list is demonic alone.
	shapes, err := s.SeriesShapes(SeriesFilter{Name: SeriesFilterNoURL, Kind: "novel"})
	if err != nil {
		t.Fatalf("SeriesShapes: %v", err)
	}
	if len(shapes) != 1 || shapes[0].Site != "demonic" ||
		shapes[0].Total != 1 || shapes[0].Novel != 1 {
		t.Fatalf("novel no-URL aggregate = %+v, want demonic {Total:1 Novel:1}", shapes)
	}
}

// The projection is the privacy boundary: a Series whose Latest Chapter was
// raised by a Reader's report reads back with the anonymous boolean set, not
// with the Reader's id, and no Reader id travels in any returned row.
func TestAdminSeriesReportsAnonymously(t *testing.T) {
	s := newTestStore(t)
	seedAdminSeries(t, s, seriesSeed{key: "asura:raised", url: "u", checkedAt: 9000, latestNum: new(9.0), bookmarks: 1, raisedBy: true})
	seedAdminSeries(t, s, seriesSeed{key: "asura:polled", url: "u", checkedAt: 9000, latestNum: new(8.0), bookmarks: 1})

	page, err := s.SeriesPage(SeriesFilter{})
	if err != nil {
		t.Fatalf("SeriesPage: %v", err)
	}
	byKey := map[string]AdminSeries{}
	for _, a := range page.Rows {
		byKey[a.Key()] = a
	}
	if !byKey["asura:raised"].RaisedByReader {
		t.Fatal("Reader-raised Series read back RaisedByReader=false")
	}
	if byKey["asura:polled"].RaisedByReader {
		t.Fatal("Poll-raised Series read back RaisedByReader=true")
	}
}

// The privacy test that cannot rot into a template-only guarantee: assert the
// admin column constant does not carry the Sighting-raiser column and that the
// admin row type has no field for it, modelled on the guard on the Bookmark
// column list.
func TestAdminProjectionHidesSightingRaiser(t *testing.T) {
	if strings.Contains(adminSeriesColumns, "latest_raised_by") {
		t.Fatal("admin column list carries latest_raised_by: the Sighting-raiser id would reach the owner")
	}
	if _, ok := reflect.TypeOf(AdminSeries{}).FieldByName("LatestRaisedBy"); ok {
		t.Fatal("AdminSeries carries a field for the Sighting-raiser id")
	}
}

// An unknown filter name is rejected rather than silently meaning "all" —
// otherwise a mistyped URL would present an empty page as the whole library.
func TestAdminFilterUnknownNameRejected(t *testing.T) {
	s := newTestStore(t)
	seedAdminSeries(t, s, seriesSeed{key: "asura:a", url: "u", checkedAt: 9000, bookmarks: 1})
	for name, call := range map[string]func() error{
		"page":  func() error { _, err := s.SeriesPage(SeriesFilter{Name: "bogus"}); return err },
		"shape": func() error { _, err := s.SeriesShapes(SeriesFilter{Name: "bogus"}); return err },
	} {
		if err := call(); err == nil || !strings.Contains(err.Error(), "unknown series filter") {
			t.Fatalf("%s with bogus filter = %v, want unknown-filter error", name, err)
		}
	}
}

// ForceSeriesPoll is the idempotent stamp write: a second press overwrites
// the request time, and touching a missing series is not an error.
func TestForceSeriesPollStampsIdempotently(t *testing.T) {
	s := newTestStore(t)
	seedAdminSeries(t, s, seriesSeed{key: "asura:x", url: "u", checkedAt: 9000, bookmarks: 1})

	if err := s.ForceSeriesPoll("asura", "x", 42); err != nil {
		t.Fatalf("ForceSeriesPoll: %v", err)
	}
	if err := s.ForceSeriesPoll("asura", "x", 99); err != nil {
		t.Fatalf("ForceSeriesPoll re-stamp: %v", err)
	}
	// Touching a missing series is not an error: the row may have been
	// orphaned, and the caller's read decides what exists.
	if err := s.ForceSeriesPoll("asura", "ghost", 99); err != nil {
		t.Fatalf("ForceSeriesPoll missing: %v", err)
	}

	var got int64
	if err := s.db.QueryRow(
		`SELECT force_poll_at FROM series WHERE site = 'asura' AND series_id = 'x'`).Scan(&got); err != nil {
		t.Fatalf("read force_poll_at: %v", err)
	}
	if got != 99 {
		t.Fatalf("force_poll_at = %d, want 99 (the later press wins)", got)
	}
}

// The admin projection carries the force stamp so the web layer can derive
// the pending flag without a second read.
func TestAdminSeriesCarriesForcePollAt(t *testing.T) {
	s := newTestStore(t)
	seedAdminSeries(t, s, seriesSeed{key: "asura:x", url: "u", checkedAt: 1000, bookmarks: 1})
	if err := s.ForceSeriesPoll("asura", "x", 5000); err != nil {
		t.Fatalf("ForceSeriesPoll: %v", err)
	}

	page, err := s.SeriesPage(SeriesFilter{})
	if err != nil {
		t.Fatalf("SeriesPage: %v", err)
	}
	if len(page.Rows) != 1 || page.Rows[0].ForcePollAt != 5000 {
		t.Fatalf("row = %+v, want ForcePollAt 5000", page.Rows)
	}
}

// SetSeriesFinished is the owner's finish stamp write: finishing writes the
// given ms, un-finishing writes zero — the one undo, the same shape as the
// correction stamp. Touching a missing series is not an error: the row may
// have been orphaned, and the caller's read decides what exists.
func TestSetSeriesFinishedStampsAndClears(t *testing.T) {
	s := newTestStore(t)
	seedAdminSeries(t, s, seriesSeed{key: "asura:x", url: "u", checkedAt: 9000, bookmarks: 1})

	if err := s.SetSeriesFinished("asura", "x", 42); err != nil {
		t.Fatalf("SetSeriesFinished: %v", err)
	}
	var got int64
	if err := s.db.QueryRow(
		`SELECT finished_at FROM series WHERE site = 'asura' AND series_id = 'x'`).Scan(&got); err != nil {
		t.Fatalf("read finished_at: %v", err)
	}
	if got != 42 {
		t.Fatalf("finished_at = %d, want 42", got)
	}

	if err := s.SetSeriesFinished("asura", "x", 0); err != nil {
		t.Fatalf("SetSeriesFinished un-finish: %v", err)
	}
	if err := s.db.QueryRow(
		`SELECT finished_at FROM series WHERE site = 'asura' AND series_id = 'x'`).Scan(&got); err != nil {
		t.Fatalf("read finished_at after un-finish: %v", err)
	}
	if got != 0 {
		t.Fatalf("finished_at = %d, want 0 (un-finish writes zero)", got)
	}

	if err := s.SetSeriesFinished("asura", "ghost", 42); err != nil {
		t.Fatalf("SetSeriesFinished missing: %v", err)
	}
}

// The admin projection carries the finish stamp so the web layer can render
// the finished state without a second read.
func TestAdminSeriesCarriesFinishedAt(t *testing.T) {
	s := newTestStore(t)
	seedAdminSeries(t, s, seriesSeed{key: "asura:x", url: "u", checkedAt: 1000, bookmarks: 1})
	if err := s.SetSeriesFinished("asura", "x", 5000); err != nil {
		t.Fatalf("SetSeriesFinished: %v", err)
	}

	page, err := s.SeriesPage(SeriesFilter{})
	if err != nil {
		t.Fatalf("SeriesPage: %v", err)
	}
	if len(page.Rows) != 1 || page.Rows[0].FinishedAt != 5000 {
		t.Fatalf("row = %+v, want FinishedAt 5000", page.Rows)
	}
}

// The failing pair reads the failure row's age, not the Series row (issue
// #165): failing requires the joined row, a Latest Chapter, and failing_since
// past the cutoff — the same constant stale reads; unverified is the
// Reader-attributed subset of the same test. A failure inside the window is
// in neither — one bad fetch is not a fault to correct — and a Series that
// never captured a chapter is never failing, the exact complement of
// never-read-a-chapter's IS NULL half, so the two filters are disjoint by
// construction. A finished failing Series still appears: this pair reads
// stored outcomes, which simply stop arriving, and carries no finished guard.
func TestAdminFailingAndUnverifiedFilters(t *testing.T) {
	s := newTestStore(t)
	const cutoff = 5000
	seedAdminSeries(t, s, seriesSeed{key: "asura:failing", url: "u", cover: "c", checkedAt: 9000, latestNum: new(10.0), bookmarks: 1})
	seedAdminSeries(t, s, seriesSeed{key: "asura:recent", url: "u", cover: "c", checkedAt: 9000, latestNum: new(9.0), bookmarks: 1})
	seedAdminSeries(t, s, seriesSeed{key: "asura:nochapter", url: "u", cover: "c", checkedAt: 9000, bookmarks: 1})
	seedAdminSeries(t, s, seriesSeed{key: "asura:reader", url: "u", cover: "c", checkedAt: 9000, latestNum: new(8.0), bookmarks: 1, raisedBy: true})
	seedAdminSeries(t, s, seriesSeed{key: "asura:polled", url: "u", cover: "c", checkedAt: 9000, latestNum: new(7.0), bookmarks: 1})
	seedAdminSeries(t, s, seriesSeed{key: "asura:fin-failing", url: "u", cover: "c", checkedAt: 9000, latestNum: new(6.0), bookmarks: 1})
	seedAdminSeries(t, s, seriesSeed{key: "asura:healthy", url: "u", cover: "c", checkedAt: 9000, latestNum: new(5.0), bookmarks: 1})

	// Failure rows seed the run's start stamp; the cutoff decides the age.
	for _, f := range []struct {
		key, word string
		since     int64
	}{
		{"asura:failing", "not_found", 2000},
		{"asura:recent", "not_found", 9000},
		{"asura:nochapter", "not_found", 2000},
		{"asura:reader", "not_found", 2000},
		{"asura:polled", "errors", 2000},
		{"asura:fin-failing", "not_found", 2000},
	} {
		site, id, _ := strings.Cut(f.key, ":")
		if err := s.RecordSeriesFailure(site, id, f.word, f.since); err != nil {
			t.Fatalf("seed failure %s: %v", f.key, err)
		}
	}
	if err := s.SetSeriesFinished("asura", "fin-failing", 1000); err != nil {
		t.Fatalf("finish asura:fin-failing: %v", err)
	}

	cases := []struct {
		name string
		f    SeriesFilter
		want []string
	}{
		{"failing", SeriesFilter{Name: SeriesFilterFailing, Cutoff: cutoff}, []string{"asura:failing", "asura:reader", "asura:polled", "asura:fin-failing"}},
		{"unverified", SeriesFilter{Name: SeriesFilterUnverified, Cutoff: cutoff}, []string{"asura:reader"}},
		{"never read a chapter", SeriesFilter{Name: SeriesFilterNoChapter}, []string{"asura:nochapter"}},
	}
	for _, tc := range cases {
		t.Run(tc.name, func(t *testing.T) {
			got := pageKeys(t, s, tc.f)
			want := map[string]bool{}
			for _, k := range tc.want {
				want[k] = true
			}
			if len(got) != len(want) {
				t.Fatalf("%+v returned %v, want exactly %v", tc.f, got, want)
			}
			for k := range want {
				if !got[k] {
					t.Fatalf("%+v dropped %q (got %v)", tc.f, k, got)
				}
			}
		})
	}

	// Disjointness: the failing pair and never-read-a-chapter are disjoint by
	// the chapter column — IS NOT NULL here, IS NULL there — so no row may be
	// counted under both.
	failing := pageKeys(t, s, SeriesFilter{Name: SeriesFilterFailing, Cutoff: cutoff})
	noChapter := pageKeys(t, s, SeriesFilter{Name: SeriesFilterNoChapter})
	for k := range failing {
		if noChapter[k] {
			t.Fatalf("row %q matches both failing and never-read-a-chapter", k)
		}
	}
	if failing["asura:nochapter"] {
		t.Fatal("a Series with no chapter ever captured appears in failing")
	}
	if !noChapter["asura:nochapter"] {
		t.Fatal("the no-chapter Series vanished from never-read-a-chapter")
	}

	// The aggregates count the same rows the lists do: the landing page's
	// figures and the select's options come from these two passes.
	for _, name := range []string{SeriesFilterFailing, SeriesFilterUnverified} {
		shapes, err := s.SeriesShapes(SeriesFilter{Name: name, Cutoff: cutoff})
		if err != nil {
			t.Fatalf("SeriesShapes(%s): %v", name, err)
		}
		sum := 0
		for _, sh := range shapes {
			sum += sh.Total
		}
		want := len(pageKeys(t, s, SeriesFilter{Name: name, Cutoff: cutoff}))
		if sum != want {
			t.Fatalf("%s aggregate sum = %d, want %d (aggregate disagrees with row query)", name, sum, want)
		}
	}
}

// The projection carries the failure facts from the LEFT JOIN, not the
// Series row: a failing Series reads back its outcome word and the stamp its
// run began at; a Series with no failure row reads empty and zero. Nothing
// new is projected from the Series row itself.
func TestAdminFailureProjection(t *testing.T) {
	s := newTestStore(t)
	seedAdminSeries(t, s, seriesSeed{key: "asura:broken", url: "u", cover: "c", checkedAt: 9000, latestNum: new(9.0), bookmarks: 1})
	seedAdminSeries(t, s, seriesSeed{key: "asura:fine", url: "u", cover: "c", checkedAt: 9000, latestNum: new(8.0), bookmarks: 1})
	if err := s.RecordSeriesFailure("asura", "broken", "not_found", 2000); err != nil {
		t.Fatalf("seed failure row: %v", err)
	}

	page, err := s.SeriesPage(SeriesFilter{})
	if err != nil {
		t.Fatalf("SeriesPage: %v", err)
	}
	byKey := map[string]AdminSeries{}
	for _, a := range page.Rows {
		byKey[a.Key()] = a
	}
	if byKey["asura:broken"].FailureOutcome != "not_found" || byKey["asura:broken"].FailingSince != 2000 {
		t.Fatalf("broken row = %+v, want FailureOutcome not_found, FailingSince 2000", byKey["asura:broken"])
	}
	if byKey["asura:fine"].FailureOutcome != "" || byKey["asura:fine"].FailingSince != 0 {
		t.Fatalf("fine row = %+v, want empty outcome and zero stamp", byKey["asura:fine"])
	}
}

// The site-completed filter (issue #170) lists the Site's own marker as work
// to work through, carrying the same finished guard the clock-driven
// predicates gained: the Site's stamp keeps standing after the owner retires
// the row, so a finished Series would be reported as work nobody is going to
// do. A zero stamp never appears. Un-finishing puts the Series back in the
// Poll query — the finished_at gate is #157's own, asserted through
// DueForLatestCheck rather than re-derived here.
func TestAdminSiteCompletedFilter(t *testing.T) {
	s := newTestStore(t)
	seedAdminSeries(t, s, seriesSeed{key: "asura:done", url: "u", cover: "c", checkedAt: 9000, latestNum: new(4.0), bookmarks: 1, siteCompletedAt: 5000})
	seedAdminSeries(t, s, seriesSeed{key: "asura:never", url: "u", cover: "c", checkedAt: 9000, latestNum: new(4.0), bookmarks: 1})
	seedAdminSeries(t, s, seriesSeed{key: "asura:retired", url: "u", cover: "c", checkedAt: 0, latestNum: new(4.0), bookmarks: 1, siteCompletedAt: 5000})
	seedAdminSeries(t, s, seriesSeed{key: "asura:healthy", url: "u", cover: "c", checkedAt: 9000, latestNum: new(4.0), bookmarks: 1})
	if err := s.SetSeriesFinished("asura", "retired", 1000); err != nil {
		t.Fatalf("finish asura:retired: %v", err)
	}

	got := pageKeys(t, s, SeriesFilter{Name: SeriesFilterSiteCompleted})
	if len(got) != 1 || !got["asura:done"] {
		t.Fatalf("site-completed returned %v, want only asura:done", got)
	}

	// The projection carries the stamp so the detail page can age it.
	page, err := s.SeriesPage(SeriesFilter{Name: SeriesFilterSiteCompleted})
	if err != nil {
		t.Fatalf("SeriesPage(site_completed): %v", err)
	}
	if len(page.Rows) != 1 || page.Rows[0].SiteCompletedAt != 5000 {
		t.Fatalf("row = %+v, want SiteCompletedAt 5000", page.Rows)
	}

	// The aggregate counts the same row: the landing figure and the select's
	// option come from this pass, so they cannot disagree with the list.
	shapes, err := s.SeriesShapes(SeriesFilter{Name: SeriesFilterSiteCompleted})
	if err != nil {
		t.Fatalf("SeriesShapes(site_completed): %v", err)
	}
	sum := 0
	for _, sh := range shapes {
		sum += sh.Total
	}
	if sum != 1 {
		t.Fatalf("site-completed aggregate = %d, want 1", sum)
	}

	// Un-finishing puts the Series back in the Poll query: the due read's
	// finished_at gate (issue #157) admits it again — asserted through the
	// Lane's own read, not re-derived here.
	due, err := s.DueForLatestCheck("asura", 1000, noCeiling)
	if err != nil {
		t.Fatalf("DueForLatestCheck: %v", err)
	}
	for _, sr := range due {
		if sr.Key() == "asura:retired" {
			t.Fatalf("a finished Series is still due for a Poll:\n%+v", due)
		}
	}
	if err := s.SetSeriesFinished("asura", "retired", 0); err != nil {
		t.Fatalf("un-finish asura:retired: %v", err)
	}
	due, err = s.DueForLatestCheck("asura", 1000, noCeiling)
	if err != nil {
		t.Fatalf("DueForLatestCheck after un-finish: %v", err)
	}
	found := false
	for _, sr := range due {
		if sr.Key() == "asura:retired" {
			found = true
		}
	}
	if !found {
		t.Fatalf("un-finished Series is not due for a Poll:\n%+v", due)
	}
}