feat(serve): wire paste fetcher + connect-time onboarding burst
5c: app.Fetcher = a per-user YouTube videoFetcher, so POST /paste mounts and resolves arbitrary-video metadata (Feature 2 goes live). 6: the connect trigger now runs an onboarding burst after discovery — summarize up to TAPIR_ONBOARD_SUMMARIZE_COUNT of the user's newest unsummarized videos via the gated Processor (Feature 1). Hard cap; explicit so it bypasses recency; every fetch still through globalFetchGate. No-op when count=0 or queue-only. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
@@ -32,7 +32,10 @@ func serialize(mu *sync.Mutex, run discoveryRunner) discoveryRunner {
|
|||||||
type discoveryTrigger struct {
|
type discoveryTrigger struct {
|
||||||
ctx context.Context
|
ctx context.Context
|
||||||
run discoveryRunner
|
run discoveryRunner
|
||||||
log *slog.Logger
|
// onboard, when set, runs after the discovery pass to summarize a capped number
|
||||||
|
// of the user's newest videos (Feature 1). Optional.
|
||||||
|
onboard func(ctx context.Context, userID string)
|
||||||
|
log *slog.Logger
|
||||||
}
|
}
|
||||||
|
|
||||||
func (t *discoveryTrigger) Enqueue(userID string) {
|
func (t *discoveryTrigger) Enqueue(userID string) {
|
||||||
@@ -40,5 +43,8 @@ func (t *discoveryTrigger) Enqueue(userID string) {
|
|||||||
if _, err := t.run(t.ctx, userID); err != nil {
|
if _, err := t.run(t.ctx, userID); err != nil {
|
||||||
t.log.Warn("discovery: connect-triggered pass had errors", "user", userID, "err", err)
|
t.log.Warn("discovery: connect-triggered pass had errors", "user", userID, "err", err)
|
||||||
}
|
}
|
||||||
|
if t.onboard != nil {
|
||||||
|
t.onboard(t.ctx, userID)
|
||||||
|
}
|
||||||
}()
|
}()
|
||||||
}
|
}
|
||||||
|
|||||||
+30
-3
@@ -210,7 +210,10 @@ func cmdServe(ctx context.Context, log *slog.Logger) error {
|
|||||||
ClientSecret: cfg.YTClientSecret,
|
ClientSecret: cfg.YTClientSecret,
|
||||||
RedirectURL: cfg.YTConnectRedirectURL,
|
RedirectURL: cfg.YTConnectRedirectURL,
|
||||||
}, secretStore, st, log)
|
}, secretStore, st, log)
|
||||||
log.Info("web youtube connect enabled", "redirect", cfg.YTConnectRedirectURL)
|
// Paste-a-URL (Feature 2): same YouTube credentials, per-user adapter built
|
||||||
|
// per request. Mounting the /paste route keys off app.Fetcher being set.
|
||||||
|
app.Fetcher = videoFetcher{cfg: cfg, secrets: secretStore}
|
||||||
|
log.Info("web youtube connect + paste enabled", "redirect", cfg.YTConnectRedirectURL)
|
||||||
}
|
}
|
||||||
|
|
||||||
// Immediate summarization for the web "Summarize" button. When the engine can
|
// Immediate summarization for the web "Summarize" button. When the engine can
|
||||||
@@ -249,9 +252,33 @@ func cmdServe(ctx context.Context, log *slog.Logger) error {
|
|||||||
// One lock shared by the scheduler and connect-triggered passes (#6) so
|
// One lock shared by the scheduler and connect-triggered passes (#6) so
|
||||||
// they never fetch concurrently — the single-fetcher invariant (ADR-018).
|
// they never fetch concurrently — the single-fetcher invariant (ADR-018).
|
||||||
runUser := serialize(&sync.Mutex{}, rawRunUser)
|
runUser := serialize(&sync.Mutex{}, rawRunUser)
|
||||||
|
// Onboarding burst (Feature 1): after the connect-triggered discovery pass,
|
||||||
|
// summarize up to OnboardSummarizeCount of the user's NEWEST unsummarized
|
||||||
|
// videos so a fresh account gets real summaries in its first session. Hard
|
||||||
|
// cap; explicit, so it bypasses the recency window — but every fetch still
|
||||||
|
// goes through globalFetchGate via the Processor. No-op when disabled
|
||||||
|
// (count 0) or queue-only (no Processor).
|
||||||
|
onboard := func(ctx context.Context, userID string) {
|
||||||
|
if cfg.OnboardSummarizeCount <= 0 || app.Processor == nil {
|
||||||
|
return
|
||||||
|
}
|
||||||
|
ids, err := st.NewestUnsummarizedVideoIDs(ctx, userID, cfg.OnboardSummarizeCount)
|
||||||
|
if err != nil {
|
||||||
|
log.Warn("onboarding: list newest unsummarized", "user", userID, "err", err)
|
||||||
|
return
|
||||||
|
}
|
||||||
|
for _, id := range ids {
|
||||||
|
if err := app.Processor.ProcessVideo(ctx, userID, id); err != nil {
|
||||||
|
log.Warn("onboarding: summarize", "user", userID, "video", id, "err", err)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
if len(ids) > 0 {
|
||||||
|
log.Info("onboarding burst complete", "user", userID, "summarized", len(ids), "cap", cfg.OnboardSummarizeCount)
|
||||||
|
}
|
||||||
|
}
|
||||||
if app.Connect != nil {
|
if app.Connect != nil {
|
||||||
app.Connect.Discovery = &discoveryTrigger{ctx: ctx, run: runUser, log: log}
|
app.Connect.Discovery = &discoveryTrigger{ctx: ctx, run: runUser, onboard: onboard, log: log}
|
||||||
log.Info("connect-triggered discovery enabled")
|
log.Info("connect-triggered discovery enabled", "onboard_cap", cfg.OnboardSummarizeCount)
|
||||||
}
|
}
|
||||||
go runScheduler(ctx, cfg.DiscoveryInterval, st, runUser, log)
|
go runScheduler(ctx, cfg.DiscoveryInterval, st, runUser, log)
|
||||||
} else {
|
} else {
|
||||||
|
|||||||
@@ -11,9 +11,29 @@ import (
|
|||||||
"gitea.d-ma.be/mathias/tapir/internal/adapters/youtube"
|
"gitea.d-ma.be/mathias/tapir/internal/adapters/youtube"
|
||||||
"gitea.d-ma.be/mathias/tapir/internal/config"
|
"gitea.d-ma.be/mathias/tapir/internal/config"
|
||||||
"gitea.d-ma.be/mathias/tapir/internal/domain"
|
"gitea.d-ma.be/mathias/tapir/internal/domain"
|
||||||
|
"gitea.d-ma.be/mathias/tapir/internal/ports"
|
||||||
"gitea.d-ma.be/mathias/tapir/internal/usecase"
|
"gitea.d-ma.be/mathias/tapir/internal/usecase"
|
||||||
|
"gitea.d-ma.be/mathias/tapir/internal/web"
|
||||||
)
|
)
|
||||||
|
|
||||||
|
// videoFetcher adapts the YouTube adapter to web.VideoFetcher for the paste flow
|
||||||
|
// (Feature 2). It builds a per-user adapter bound to that user's token ref and
|
||||||
|
// resolves a single video's metadata via the Data API — ungated; only the later
|
||||||
|
// transcript fetch goes through globalFetchGate.
|
||||||
|
type videoFetcher struct {
|
||||||
|
cfg config.Config
|
||||||
|
secrets ports.SecretStore
|
||||||
|
}
|
||||||
|
|
||||||
|
func (f videoFetcher) FetchVideo(ctx context.Context, userID, videoID string) (domain.Video, error) {
|
||||||
|
a := youtube.New(youtube.Config{
|
||||||
|
ClientID: f.cfg.YTClientID,
|
||||||
|
ClientSecret: f.cfg.YTClientSecret,
|
||||||
|
TokenSecretRef: web.YouTubeTokenRef(userID),
|
||||||
|
}, f.secrets)
|
||||||
|
return a.VideoByID(ctx, userID, videoID)
|
||||||
|
}
|
||||||
|
|
||||||
// buildProcessor wires the summarization engine — YouTube source (captions-first),
|
// buildProcessor wires the summarization engine — YouTube source (captions-first),
|
||||||
// AI-router summarizer, store sink — shared by `tapir run` and the web
|
// AI-router summarizer, store sink — shared by `tapir run` and the web
|
||||||
// "Summarize now" path so the wiring lives in one place. It returns (nil, nil) —
|
// "Summarize now" path so the wiring lives in one place. It returns (nil, nil) —
|
||||||
|
|||||||
Reference in New Issue
Block a user