Compare commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
1aa8a97f95 | ||
|
|
cc69a912f4 | ||
|
|
f4a0544903 | ||
|
|
e2a52789b9 | ||
|
|
e696b6405b |
@@ -86,7 +86,7 @@ Skills live in the canonical library `mathias/skills` and are wired into this re
|
||||
|
||||
## Current build state (start here for the first task)
|
||||
|
||||
The repo is **green and shipping** — last tag `v0.14.0`. `task check` passes (fmt, vet, lint,
|
||||
The repo is **green and shipping** — last tag `v0.15.0`. `task check` passes (fmt, vet, lint,
|
||||
`go test -p 1 ./...`). Go is `1.26.1` (see `go.mod`).
|
||||
|
||||
- Clean Architecture core is implemented: `internal/domain` (entities), `internal/ports`
|
||||
@@ -95,7 +95,8 @@ The repo is **green and shipping** — last tag `v0.14.0`. `task check` passes (
|
||||
green against it.
|
||||
- Adapters present under `internal/adapters/`: `youtube` (captions-first `VideoSource`,
|
||||
timedtext/InnerTube acquisition per ADR-010), `summarizer` + `llm` (the copied AI router,
|
||||
Primary→Fallback per ADR-004), `store` (Postgres, golang-migrate migrations 001–015),
|
||||
now a resilient endpoint chain — local primary → local fallback → external worst-case,
|
||||
parse-failure-aware, ADR-004 + ADR-022), `store` (Postgres, golang-migrate migrations 001–015),
|
||||
`secrets` (file-backed `SecretStore`). The brain HTTP sink (ADR-005) is the remaining
|
||||
optional sink.
|
||||
- Stage 1 is open (ADR-012): multi-user with **DB-enforced** isolation — Postgres RLS `FORCE`d
|
||||
|
||||
+54
-8
@@ -36,6 +36,11 @@ func buildUserRunner(cfg config.Config, st *store.Store, secretStore ports.Secre
|
||||
}, secretStore)
|
||||
|
||||
engine := usecase.NewEngine(src, buildSummarizer(cfg), st)
|
||||
// Share the transcript cache (ADR-021) on the scheduler path too — without
|
||||
// this every scheduled pass re-fetches transcripts it already had, burning the
|
||||
// scarce per-IP caption budget (ADR-014) and starving other users. The web
|
||||
// "Summarize now" path already sets this; the scheduler omitting it was a bug.
|
||||
engine.Transcripts = st
|
||||
|
||||
return runner.New(src, st, engine, userID, log,
|
||||
runner.WithBackoff(cfg.FetchBackoff),
|
||||
@@ -57,6 +62,7 @@ type userLister interface {
|
||||
// isolation). Returns the stats summed across users.
|
||||
func runDiscoveryPass(
|
||||
ctx context.Context,
|
||||
pass int,
|
||||
lister userLister,
|
||||
runUser func(context.Context, string) (runner.Stats, error),
|
||||
log *slog.Logger,
|
||||
@@ -67,15 +73,17 @@ func runDiscoveryPass(
|
||||
return runner.Stats{}
|
||||
}
|
||||
|
||||
log.Info("scheduler: starting discovery pass", "users", len(users))
|
||||
var total runner.Stats
|
||||
// Keep only users with a video connection. A pass for a connectionless user
|
||||
// (e.g. a stale Dex-era orphan identity) only tries to resolve a token that
|
||||
// was never minted, logging a spurious "ref not found" every tick. Filtering
|
||||
// here — BEFORE rotation — also keeps fairness honest: rotation is over the
|
||||
// users that actually consume the caption budget, so a dead identity can't eat
|
||||
// a rotation slot and skew the lead share.
|
||||
var connected []store.UserIdentity
|
||||
for _, u := range users {
|
||||
if ctx.Err() != nil {
|
||||
break // shutting down: stop enumerating
|
||||
return runner.Stats{} // shutting down
|
||||
}
|
||||
// Skip users with no video connection. A discovery pass for them only
|
||||
// attempts to resolve a token that was never minted, logging a spurious
|
||||
// "ref not found" every tick (e.g. stale Dex-era orphan identities).
|
||||
conns, err := lister.ConnectionsForUser(ctx, u.UserID)
|
||||
if err != nil {
|
||||
log.Warn("scheduler: list connections failed", "user", u.UserID, "err", err)
|
||||
@@ -85,6 +93,22 @@ func runDiscoveryPass(
|
||||
log.Debug("scheduler: skipping user with no video connections", "user", u.UserID)
|
||||
continue
|
||||
}
|
||||
connected = append(connected, u)
|
||||
}
|
||||
|
||||
// Rotate who goes first each pass. Caption fetches share one per-egress-IP
|
||||
// rate budget (ADR-014); whoever runs first each pass spends the pre-throttle
|
||||
// window, so a FIXED order permanently starves whoever is last (a new pilot
|
||||
// user got 0 fetches for 12h while the first-listed user got all of them).
|
||||
// Rotation over the connected set gives each real user the lead in turn.
|
||||
connected = rotateUsers(connected, pass)
|
||||
|
||||
log.Info("scheduler: starting discovery pass", "users", len(connected))
|
||||
var total runner.Stats
|
||||
for _, u := range connected {
|
||||
if ctx.Err() != nil {
|
||||
break // shutting down: stop enumerating
|
||||
}
|
||||
stats, err := runUser(ctx, u.UserID)
|
||||
total = sumStats(total, stats)
|
||||
if err != nil {
|
||||
@@ -116,7 +140,8 @@ func runScheduler(
|
||||
return // disabled
|
||||
}
|
||||
|
||||
runDiscoveryPass(ctx, lister, runUser, log)
|
||||
pass := 0
|
||||
runDiscoveryPass(ctx, pass, lister, runUser, log)
|
||||
|
||||
ticker := time.NewTicker(interval)
|
||||
defer ticker.Stop()
|
||||
@@ -125,11 +150,32 @@ func runScheduler(
|
||||
case <-ctx.Done():
|
||||
return
|
||||
case <-ticker.C:
|
||||
runDiscoveryPass(ctx, lister, runUser, log)
|
||||
pass++
|
||||
runDiscoveryPass(ctx, pass, lister, runUser, log)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// rotateUsers left-rotates users by pass positions so a different user leads each
|
||||
// pass. With n users, user i leads on every pass where pass ≡ i (mod n). A pass
|
||||
// offset that is negative or exceeds n is normalised. Order within the rotation
|
||||
// is otherwise preserved, so the set of users run is unchanged — only who is
|
||||
// first (and thus wins the scarce caption-fetch budget) rotates.
|
||||
func rotateUsers(users []store.UserIdentity, pass int) []store.UserIdentity {
|
||||
n := len(users)
|
||||
if n <= 1 {
|
||||
return users
|
||||
}
|
||||
off := ((pass % n) + n) % n
|
||||
if off == 0 {
|
||||
return users
|
||||
}
|
||||
out := make([]store.UserIdentity, 0, n)
|
||||
out = append(out, users[off:]...)
|
||||
out = append(out, users[:off]...)
|
||||
return out
|
||||
}
|
||||
|
||||
// sumStats adds two passes' stats field-wise, so runDiscoveryPass can report a
|
||||
// per-tick aggregate across all users.
|
||||
func sumStats(a, b runner.Stats) runner.Stats {
|
||||
|
||||
@@ -45,6 +45,7 @@ func (f fakeLister) ConnectionsForUser(_ context.Context, userID string) ([]stor
|
||||
type countingRunUser struct {
|
||||
mu sync.Mutex
|
||||
calls map[string]int
|
||||
order []string // userIDs in the order they were run, across all passes
|
||||
failFor map[string]bool
|
||||
}
|
||||
|
||||
@@ -60,12 +61,19 @@ func (c *countingRunUser) run(_ context.Context, userID string) (runner.Stats, e
|
||||
c.mu.Lock()
|
||||
defer c.mu.Unlock()
|
||||
c.calls[userID]++
|
||||
c.order = append(c.order, userID)
|
||||
if c.failFor[userID] {
|
||||
return runner.Stats{Errors: 1}, errors.New("boom")
|
||||
}
|
||||
return runner.Stats{Summarized: 1}, nil
|
||||
}
|
||||
|
||||
func (c *countingRunUser) runOrder() []string {
|
||||
c.mu.Lock()
|
||||
defer c.mu.Unlock()
|
||||
return append([]string(nil), c.order...)
|
||||
}
|
||||
|
||||
func (c *countingRunUser) count(userID string) int {
|
||||
c.mu.Lock()
|
||||
defer c.mu.Unlock()
|
||||
@@ -94,7 +102,7 @@ func TestDiscoveryPassRunsEveryUserOnce(t *testing.T) {
|
||||
lister := fakeLister{users: usersN("a", "b", "c")}
|
||||
rc := newCountingRunUser()
|
||||
|
||||
stats := runDiscoveryPass(context.Background(), lister, rc.run, quietLog())
|
||||
stats := runDiscoveryPass(context.Background(), 0, lister, rc.run, quietLog())
|
||||
|
||||
require.Equal(t, 1, rc.count("a"))
|
||||
require.Equal(t, 1, rc.count("b"))
|
||||
@@ -102,13 +110,46 @@ func TestDiscoveryPassRunsEveryUserOnce(t *testing.T) {
|
||||
require.Equal(t, 3, stats.Summarized, "stats are summed across users")
|
||||
}
|
||||
|
||||
// Caption fetches share one per-IP budget; a fixed user order starves whoever is
|
||||
// last. Each pass must rotate which user leads so the lead slot is shared.
|
||||
func TestDiscoveryPassRotatesLeadUser(t *testing.T) {
|
||||
lister := fakeLister{users: usersN("a", "b", "c")}
|
||||
rc := newCountingRunUser()
|
||||
|
||||
runDiscoveryPass(context.Background(), 0, lister, rc.run, quietLog())
|
||||
runDiscoveryPass(context.Background(), 1, lister, rc.run, quietLog())
|
||||
runDiscoveryPass(context.Background(), 2, lister, rc.run, quietLog())
|
||||
|
||||
require.Equal(t, []string{"a", "b", "c", "b", "c", "a", "c", "a", "b"}, rc.runOrder(),
|
||||
"each pass left-rotates the user order so every user leads in turn")
|
||||
// Fairness: over a full rotation cycle every user ran the same number of times.
|
||||
require.Equal(t, 3, rc.count("a"))
|
||||
require.Equal(t, 3, rc.count("b"))
|
||||
require.Equal(t, 3, rc.count("c"))
|
||||
}
|
||||
|
||||
// A connectionless orphan must not consume a rotation slot: rotation is over the
|
||||
// connected users only, so two real users alternate the lead 50/50 even with a
|
||||
// dead identity listed between them.
|
||||
func TestDiscoveryPassRotationIgnoresConnectionlessUsers(t *testing.T) {
|
||||
lister := fakeLister{users: usersN("a", "orphan", "c"), noConn: map[string]bool{"orphan": true}}
|
||||
rc := newCountingRunUser()
|
||||
|
||||
runDiscoveryPass(context.Background(), 0, lister, rc.run, quietLog())
|
||||
runDiscoveryPass(context.Background(), 1, lister, rc.run, quietLog())
|
||||
|
||||
require.Equal(t, []string{"a", "c", "c", "a"}, rc.runOrder(),
|
||||
"only connected users rotate; the orphan never runs and never holds a slot")
|
||||
require.Equal(t, 0, rc.count("orphan"))
|
||||
}
|
||||
|
||||
func TestDiscoveryPassSkipsUsersWithoutConnections(t *testing.T) {
|
||||
// b never connected a video source (e.g. a stale Dex-era orphan identity).
|
||||
// It must be skipped silently — not run and logged as a token error every pass.
|
||||
lister := fakeLister{users: usersN("a", "b", "c"), noConn: map[string]bool{"b": true}}
|
||||
rc := newCountingRunUser()
|
||||
|
||||
stats := runDiscoveryPass(context.Background(), lister, rc.run, quietLog())
|
||||
stats := runDiscoveryPass(context.Background(), 0, lister, rc.run, quietLog())
|
||||
|
||||
require.Equal(t, 1, rc.count("a"))
|
||||
require.Equal(t, 0, rc.count("b"), "a user with no connection must be skipped, not run")
|
||||
@@ -121,7 +162,7 @@ func TestDiscoveryPassOneUserFailureDoesNotStopOthers(t *testing.T) {
|
||||
lister := fakeLister{users: usersN("a", "b", "c")}
|
||||
rc := newCountingRunUser("b") // user b's pass errors
|
||||
|
||||
stats := runDiscoveryPass(context.Background(), lister, rc.run, quietLog())
|
||||
stats := runDiscoveryPass(context.Background(), 0, lister, rc.run, quietLog())
|
||||
|
||||
require.Equal(t, 1, rc.count("a"))
|
||||
require.Equal(t, 1, rc.count("b"))
|
||||
@@ -134,7 +175,7 @@ func TestDiscoveryPassListerErrorIsContained(t *testing.T) {
|
||||
lister := fakeLister{err: errors.New("db down")}
|
||||
rc := newCountingRunUser()
|
||||
|
||||
stats := runDiscoveryPass(context.Background(), lister, rc.run, quietLog())
|
||||
stats := runDiscoveryPass(context.Background(), 0, lister, rc.run, quietLog())
|
||||
|
||||
require.Equal(t, 0, rc.total(), "no users enumerated → no passes")
|
||||
require.Equal(t, runner.Stats{}, stats)
|
||||
|
||||
@@ -233,11 +233,20 @@ func (a *App) handleList(w http.ResponseWriter, r *http.Request) {
|
||||
}
|
||||
hasConnected := len(conns) > 0
|
||||
|
||||
if isHTMX(r) {
|
||||
a.render(w, r, summaryList(buckets, hasConnected))
|
||||
// Summarization mode drives the backlog copy: an auto user is told summaries
|
||||
// land gradually; a manual user is told to click Summarize (the first pilot
|
||||
// user sat in manual mode reading "land automatically" and waited forever).
|
||||
autoSummarize, err := a.Store.GetAutoSummarize(r.Context(), userID)
|
||||
if err != nil {
|
||||
a.serverError(w, r, "summarize mode", err)
|
||||
return
|
||||
}
|
||||
a.render(w, r, ListPage(buckets, f, stats, takeFlash(w, r), hasConnected, channels))
|
||||
|
||||
if isHTMX(r) {
|
||||
a.render(w, r, summaryList(buckets, hasConnected, autoSummarize))
|
||||
return
|
||||
}
|
||||
a.render(w, r, ListPage(buckets, f, stats, takeFlash(w, r), hasConnected, channels, autoSummarize))
|
||||
}
|
||||
|
||||
// handleDetail renders one summary in full (highlights, takeaways, action group).
|
||||
|
||||
@@ -477,3 +477,36 @@ func postAction(t *testing.T, app *web.App, videoID, action string, htmx bool) *
|
||||
}
|
||||
return do(t, app, req)
|
||||
}
|
||||
|
||||
// TestListManualModeBannerCopy: a manual-mode user with un-summarized videos
|
||||
// sees the manual prompt (click Summarize), NOT the "summaries land
|
||||
// automatically" copy that misled the first pilot user into waiting forever.
|
||||
func TestListManualModeBannerCopy(t *testing.T) {
|
||||
ctx := context.Background()
|
||||
app := newApp(t)
|
||||
p := rawPool(t)
|
||||
resetDB(t, p)
|
||||
seedVideo(t, p, videoX, "X Title", "https://x", time.Time{}) // pending, un-summarized
|
||||
require.NoError(t, app.Store.SetAutoSummarize(ctx, userID, false))
|
||||
|
||||
html := body(t, do(t, app, httptest.NewRequest(http.MethodGet, "/", nil)))
|
||||
require.Contains(t, html, "Manual mode")
|
||||
require.Contains(t, html, "are not summarized automatically")
|
||||
require.NotContains(t, html, "land gradually",
|
||||
"manual-mode user must not be told summaries arrive automatically")
|
||||
}
|
||||
|
||||
// TestListAutoModeBannerCopy: an auto-mode user with a backlog sees the
|
||||
// gradual-delivery copy, not the manual prompt.
|
||||
func TestListAutoModeBannerCopy(t *testing.T) {
|
||||
ctx := context.Background()
|
||||
app := newApp(t)
|
||||
p := rawPool(t)
|
||||
resetDB(t, p)
|
||||
seedVideo(t, p, videoX, "X Title", "https://x", time.Time{})
|
||||
require.NoError(t, app.Store.SetAutoSummarize(ctx, userID, true))
|
||||
|
||||
html := body(t, do(t, app, httptest.NewRequest(http.MethodGet, "/", nil)))
|
||||
require.Contains(t, html, "land gradually")
|
||||
require.NotContains(t, html, "are not summarized automatically")
|
||||
}
|
||||
|
||||
@@ -65,7 +65,7 @@ func TestParseYouTubeVideoID(t *testing.T) {
|
||||
func TestListPageShowsPasteFormOnlyWhenConnected(t *testing.T) {
|
||||
render := func(connected bool) string {
|
||||
var buf bytes.Buffer
|
||||
if err := ListPage(listBuckets{}, Filter{}, PipelineStats{}, "", connected, nil).Render(context.Background(), &buf); err != nil {
|
||||
if err := ListPage(listBuckets{}, Filter{}, PipelineStats{}, "", connected, nil, true).Render(context.Background(), &buf); err != nil {
|
||||
t.Fatalf("render: %v", err)
|
||||
}
|
||||
return buf.String()
|
||||
|
||||
@@ -104,7 +104,7 @@ templ flashBanner(code string) {
|
||||
// #summary-list region; a non-HTMX request renders the whole page. flash carries
|
||||
// a one-shot notification (e.g. "connected", "registered") surfaced on arrival
|
||||
// after a POST→redirect.
|
||||
templ ListPage(b listBuckets, f Filter, stats PipelineStats, flash string, hasConnected bool, channels []string) {
|
||||
templ ListPage(b listBuckets, f Filter, stats PipelineStats, flash string, hasConnected bool, channels []string, autoSummarize bool) {
|
||||
@Layout("Tapir — Summaries") {
|
||||
@flashBanner(flash)
|
||||
if hasConnected {
|
||||
@@ -116,14 +116,23 @@ templ ListPage(b listBuckets, f Filter, stats PipelineStats, flash string, hasCo
|
||||
if stats.RateLimited > 0 || stats.Pending > 0 || stats.NoText > 0 {
|
||||
@pipelineBar(stats)
|
||||
}
|
||||
if stats.RateLimited+stats.Pending > 0 {
|
||||
if (stats.RateLimited+stats.Pending) > 0 && autoSummarize {
|
||||
<p class="pipeline-note muted">
|
||||
Tapir fetches captions slowly on purpose, to respect YouTube's limits —
|
||||
new summaries land gradually. Check back tomorrow.
|
||||
</p>
|
||||
}
|
||||
if (stats.RateLimited+stats.Pending) > 0 && !autoSummarize {
|
||||
<p class="pipeline-note muted">
|
||||
You are in Manual mode: new videos appear here but are not summarized
|
||||
automatically. Use the Summarize button on the ones you want.
|
||||
</p>
|
||||
<p class="pipeline-note muted">
|
||||
<a href="/account">Switch to Automatic</a> to have new videos summarized for you.
|
||||
</p>
|
||||
}
|
||||
<div id="summary-list">
|
||||
@summaryList(b, hasConnected)
|
||||
@summaryList(b, hasConnected, autoSummarize)
|
||||
</div>
|
||||
}
|
||||
}
|
||||
@@ -203,12 +212,16 @@ templ filterForm(f Filter, channels []string) {
|
||||
// and a single disclosure holding the older un-summarized back-catalogue. Cards
|
||||
// reflow to a single column on mobile; an empty list shows a friendly first-run
|
||||
// state instead of a blank table.
|
||||
templ summaryList(b listBuckets, hasConnected bool) {
|
||||
templ summaryList(b listBuckets, hasConnected bool, autoSummarize bool) {
|
||||
if b.empty() {
|
||||
if hasConnected {
|
||||
<div class="empty empty-connected">
|
||||
<strong>Your account is connected</strong>
|
||||
if autoSummarize {
|
||||
<span>Tapir is finding your subscriptions and fetching captions — summaries appear here gradually. Check back later.</span>
|
||||
} else {
|
||||
<span>Tapir is finding your subscriptions. You are in Manual mode, so videos appear here with a Summarize button — pick the ones you want, or switch to Automatic in your account.</span>
|
||||
}
|
||||
</div>
|
||||
} else {
|
||||
<div class="empty">
|
||||
|
||||
+274
-249
File diff suppressed because it is too large
Load Diff
Reference in New Issue
Block a user