package statedb

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"time"

	"github.com/bluesky-social/indigo/atproto/syntax"
	"github.com/bluesky-social/indigo/util"
	"github.com/bluesky-social/indigo/xrpc"
	glex "github.com/streamplace/glex/runtime"
	"golang.org/x/sync/errgroup"
	"gorm.io/gorm"
	"stream.place/streamplace/pkg/appbsky"
	"stream.place/streamplace/pkg/integrations/webhook"
	"stream.place/streamplace/pkg/log"
	"stream.place/streamplace/pkg/model"
	notificationpkg "stream.place/streamplace/pkg/notifications"
	"stream.place/streamplace/pkg/placestream"

	"stream.place/streamplace/pkg/comatproto"
)

var TaskNotification = "notification"
var TaskChat = "chat"
var TaskStreamReceived = "stream_received"
var TaskFinalizeLivestream = "finalize_livestream"
var TaskFinalizeLivestreamVOD = "finalize_livestream_vod"
var TaskVODProcess = "vod_process"
var TaskViewCountAggregate = "view_count_aggregate"
var TaskCDNLogIngest = "cdn_log_ingest"

// nonVODTaskTypes is every task type handled by the general queue worker.
// VOD processing runs on its own dedicated pool (see ProcessQueue) so a
// slow remux can't block these lighter tasks, so VOD is deliberately
// excluded here. Keep this list in sync when adding a new task type that
// is NOT VOD — anything missing from both this list and the VOD pool will
// never be dequeued.
var nonVODTaskTypes = []string{
	TaskNotification,
	TaskChat,
	TaskStreamReceived,
	TaskFinalizeLivestream,
	TaskViewCountAggregate,
	TaskCDNLogIngest,
}

type NotificationTask struct {
	Livestream  placestream.Livestream_LivestreamView
	FeedPost    *appbsky.FeedDefs_PostView
	ChatProfile placestream.ChatProfile
	PDSURL      string
}

type ChatTask struct {
	MessageView placestream.ChatDefs_MessageView
}

type StreamReceivedTask struct {
	StreamerDID string `json:"streamerDID"`
}

type FinalizeLivestreamTask struct {
	LivestreamURI string `json:"livestreamURI"`
	// StaleSince is the record's lastSeenAt as it stood when this task was
	// rescheduled to give the heartbeat one more idle window. If it is still
	// that when the task runs again, the heartbeat is frozen and the record
	// is ended; without it the task rescheduled itself forever for a repo's
	// latest un-ended record.
	StaleSince string `json:"staleSince,omitempty"`
}

// heartbeatFrozen reports whether a rescheduled finalize found the record's
// heartbeat exactly where it left it a full idle window ago.
func heartbeatFrozen(task FinalizeLivestreamTask, lastSeenAt string) bool {
	return task.StaleSince != "" && task.StaleSince == lastSeenAt
}

// VODProcessTask is enqueued by the upload manager when a resumable user
// upload completes. The processor probes the file, generates MUXL tracks,
// and creates the place.stream.video record set. The actual processing is
// not yet implemented — for now this just acknowledges receipt.
type VODProcessTask struct {
	UploadID string `json:"uploadId"`
	RepoDID  string `json:"repoDID"`
	MimeType string `json:"mimeType"`
	Filename string `json:"filename,omitempty"`
	Size     int64  `json:"size"`
	Backend  string `json:"backend"`
	Location string `json:"location"`
}

// FinalizeLivestreamVODTask is enqueued by the place.stream.media.finalizeLivestream
// procedure to turn a finished livestream's recorded MUXL objects into a VOD.
// UploadID is the synthetic Upload row the client polls (getUploadStatus) and
// then publishes (publishVideo), exactly as for a resumable upload.
type FinalizeLivestreamVODTask struct {
	UploadID      string `json:"uploadId"`
	RepoDID       string `json:"repoDID"`
	LivestreamURI string `json:"livestreamURI"`
	// LivestreamURIs, when set, are all the livestream records whose
	// recordings make up the VOD, in order (LivestreamURI is the first);
	// empty means just LivestreamURI.
	LivestreamURIs []string `json:"livestreamURIs,omitempty"`
	// Publish, when set, describes the place.stream.video record to
	// publish in the streamer's repo (with their stored session) as soon
	// as the VOD is finalized, instead of leaving a draft for them to
	// publish from the app. Set by the operator's finalize route.
	Publish *VideoDraft `json:"publish,omitempty"`
}

// VideoDraft is the part of a place.stream.video record known before the
// VOD exists: what the operator (or the livestream record) says about it.
// It is not a placestream.Video because that record's source union has no
// value until the finalizer has the MUXL CID, and an empty union will not
// marshal, so a Video cannot sit in a task payload; the publisher builds
// the record from this with Record() and fills in the source, duration and
// thumbnail from the finished upload.
type VideoDraft struct {
	Title       string                               `json:"title"`
	Description *string                              `json:"description,omitempty"`
	Tags        []string                             `json:"tags,omitempty"`
	Activity    *placestream.Video_Activity          `json:"activity,omitempty"`
	Connections []placestream.Video_Connections_Elem `json:"connections,omitempty"`
}

// Record is the place.stream.video record for the draft, minus the fields
// that come from the finalized upload.
func (d *VideoDraft) Record() *placestream.Video {
	return &placestream.Video{
		LexiconTypeID: "place.stream.video",
		Title:         d.Title,
		Description:   d.Description,
		Tags:          d.Tags,
		Activity:      d.Activity,
		Connections:   d.Connections,
	}
}

// VideoPublisher publishes a finalized livestream VOD's place.stream.video
// record; installed at bootstrap like LivestreamVODFinalizer.
type VideoPublisher func(ctx context.Context, t FinalizeLivestreamVODTask) (uri, cid string, err error)

func (state *StatefulDB) SetVideoPublisher(f VideoPublisher) {
	state.videoPublisher = f
}

// ViewCountAggregateTask is the payload for one aggregation window.
// Enqueued by every streamplace node at the configured interval; the
// unique task key (built from WindowStart/End) ensures only one node's
// enqueue + dequeue actually runs each window.
type ViewCountAggregateTask struct {
	WindowStart time.Time `json:"windowStart"`
	WindowEnd   time.Time `json:"windowEnd"`
}

// CDNLogIngestTask is the payload for one scheduled pass over the
// CDN's archived access logs. Tick is the UTC-aligned schedule slot;
// it's in the dedup key, not used by the ingester itself (which
// always looks for whatever is new).
type CDNLogIngestTask struct {
	Tick time.Time `json:"tick"`
}

// ProcessQueue runs the task queue until ctx is cancelled. VOD tasks are
// handled by a dedicated pool of vodConcurrency workers, so a slow remux
// can't block quick uploads behind it (or starve the lighter tasks);
// everything else runs on a single general worker. DequeueTask uses
// FOR UPDATE SKIP LOCKED on Postgres, so the workers never claim the same
// row. vodConcurrency is clamped to at least 1.
func (state *StatefulDB) ProcessQueue(ctx context.Context, vodConcurrency int) error {
	if vodConcurrency < 1 {
		vodConcurrency = 1
	}
	log.Log(ctx, "starting task queue", "vod_concurrency", vodConcurrency)

	group, ctx := errgroup.WithContext(ctx)

	// General worker: everything except VOD.
	group.Go(func() error {
		return state.runQueueWorker(ctx, "queue_processor", nonVODTaskTypes)
	})

	// Dedicated VOD pool. Live-to-VOD finalize also runs here: it's heavy
	// I/O (a full read of the recorded stream to hash + index it) that would
	// otherwise hog the single general worker and stall light tasks.
	for i := 0; i < vodConcurrency; i++ {
		workerID := fmt.Sprintf("vod_worker_%d", i)
		group.Go(func() error {
			return state.runQueueWorker(ctx, workerID, []string{TaskVODProcess, TaskFinalizeLivestreamVOD})
		})
	}

	return group.Wait()
}

// runQueueWorker pulls and processes tasks of the given types until ctx is
// cancelled. A failed task is only logged here; it stays locked until its
// lease expires and then becomes eligible for retry (capped by max_tries),
// matching the queue's existing retry semantics. Empty dequeues back off
// on a 1s timer or a queue poke.
func (state *StatefulDB) runQueueWorker(ctx context.Context, workerID string, taskTypes []string) error {
	for {
		task, err := state.DequeueTask(ctx, workerID, taskTypes...)
		if err != nil && !errors.Is(err, gorm.ErrRecordNotFound) {
			return err
		}
		if task != nil {
			// Defense-in-depth: a panic inside a task handler (e.g. a
			// nil-deref in a codec, or a malformed record) must not crash
			// this worker goroutine — errgroup would cancel the whole queue.
			// Recover it into a logged error so the task fails and its retry
			// semantics (lease expiry + max_tries) still apply.
			if err := log.Recover(ctx, func() error {
				return state.processTask(ctx, task)
			}); err != nil {
				log.Error(ctx, "failed to process task", "err", err, "worker", workerID, "taskId", fmt.Sprintf("%d", task.ID))
			}
			continue
		}
		select {
		case <-ctx.Done():
			return ctx.Err()
		case <-time.After(1 * time.Second):
		case <-state.pokeQueue:
		}
	}
}

func (state *StatefulDB) processTask(ctx context.Context, task *AppTask) error {
	ctx = log.WithLogValues(ctx, "taskType", task.Type, "taskId", fmt.Sprintf("%d", task.ID))
	switch task.Type {
	case TaskNotification:
		return state.processNotificationTask(ctx, task)
	case TaskChat:
		return state.processChatMessageTask(ctx, task)
	case TaskStreamReceived:
		return state.processStreamReceivedTask(ctx, task)
	case TaskFinalizeLivestream:
		return state.processFinalizeLivestreamTask(ctx, task)
	case TaskVODProcess:
		return state.processVODProcessTask(ctx, task)
	case TaskFinalizeLivestreamVOD:
		return state.processFinalizeLivestreamVODTask(ctx, task)
	case TaskViewCountAggregate:
		return state.processViewCountAggregateTask(ctx, task)
	case TaskCDNLogIngest:
		return state.processCDNLogIngestTask(ctx, task)
	default:
		return fmt.Errorf("unknown task type: %s", task.Type)
	}
}

// VODProcessor runs the gstreamer + muxl + S3 pipeline for one upload
// and returns the resulting BDASL CID. The function-pointer indirection
// keeps pkg/statedb from importing pkg/vod (which transitively pulls in
// gstreamer); the bootstrap (pkg/cmd) installs the concrete
// implementation at startup.
type VODProcessor func(ctx context.Context, t VODProcessTask) (cid string, err error)

func (state *StatefulDB) SetVODProcessor(f VODProcessor) { state.vodProcessor = f }

// SetNotifier installs the notification notifier after construction. This is
// needed because building the Web Push notifier requires VAPID keys, which
// are stored in the DB — so the DB must exist before the notifier can be
// fully assembled. The queue processor checks for nil, so a brief window
// with no notifier is safe.
func (state *StatefulDB) SetNotifier(n notificationpkg.Notifier) { state.noter = n }

func (state *StatefulDB) processVODProcessTask(ctx context.Context, task *AppTask) error {
	ctx = log.WithLogValues(ctx, "func", "processVODProcessTask")
	var t VODProcessTask
	if err := json.Unmarshal(task.Payload, &t); err != nil {
		return err
	}
	// Thread the upload + owner into the log context as early as possible so
	// every downstream log line (and the error returned below, which is
	// re-logged by runQueueWorker with the loop's context) carries the user.
	ctx = log.WithLogValues(ctx, "uploadId", t.UploadID, "did", t.RepoDID)
	log.Log(ctx, "dequeued vod-process task")
	if state.vodProcessor == nil {
		log.Warn(ctx, "no VOD processor configured; dropping task",
			"uploadId", t.UploadID, "did", t.RepoDID)
		return state.CompleteTask(ctx, task.ID)
	}
	if err := state.SetUploadProcessing(ctx, t.UploadID); err != nil {
		log.Warn(ctx, "failed to mark upload as processing", "uploadId", t.UploadID, "error", err)
	}
	cid, err := state.vodProcessor(ctx, t)
	if err != nil {
		if ferr := state.SetUploadFailed(ctx, t.UploadID, err.Error()); ferr != nil {
			log.Warn(ctx, "failed to mark upload as failed", "uploadId", t.UploadID, "error", ferr)
		}
		// Flip any tied draft to 'error' too, so the user sees the failure in
		// the Drafts tab instead of an indefinite 'processing' state.
		if derr := state.SetDraftError(ctx, t.UploadID, err.Error()); derr != nil {
			log.Warn(ctx, "failed to mark draft as failed", "uploadId", t.UploadID, "error", derr)
		}
		// Complete the task so it doesn't retry — most VOD failures are
		// permanent (unsupported codec, corrupted file, etc.).
		_ = state.CompleteTask(ctx, task.ID)
		// The upload ID + DID are now in the log context (set above), so the
		// error string no longer needs to embed them for traceability — the
		// runQueueWorker re-log picks them up from context.
		return fmt.Errorf("vod processing upload %s: %w", t.UploadID, err)
	}
	// The processor (vod.ProcessVOD) calls SetUploadProcessed deep inside its
	// own publish path, so by the time it returns the Upload row carries the
	// finished TrackURIs / DurationMS / ContentCID. Re-read them and flip the
	// tied draft to 'ready'. A missing draft (pre-drafts-era upload, or one
	// whose create failed) is a no-op, not an error.
	if err := state.markDraftReadyFromUpload(ctx, t.UploadID); err != nil {
		log.Warn(ctx, "failed to mark draft ready", "uploadId", t.UploadID, "error", err)
	}
	log.Log(ctx, "vod processed", "uploadId", t.UploadID, "cid", cid)
	return state.CompleteTask(ctx, task.ID)
}

// LivestreamVODFinalizer concatenates a finished livestream's recorded MUXL
// objects into a content-addressed VOD blob, derives its sidecars, and
// publishes the origin + track records. Returns the resulting BDASL CID. Same
// function-pointer indirection as VODProcessor so pkg/statedb doesn't import
// the blob.Store/muxl-heavy pkg/vod.
type LivestreamVODFinalizer func(ctx context.Context, t FinalizeLivestreamVODTask) (cid string, err error)

func (state *StatefulDB) SetLivestreamVODFinalizer(f LivestreamVODFinalizer) {
	state.livestreamVODFinalizer = f
}

func (state *StatefulDB) processFinalizeLivestreamVODTask(ctx context.Context, task *AppTask) error {
	ctx = log.WithLogValues(ctx, "func", "processFinalizeLivestreamVODTask")
	var t FinalizeLivestreamVODTask
	if err := json.Unmarshal(task.Payload, &t); err != nil {
		return err
	}
	// Thread the upload + owner into the log context so every downstream log
	// line (and the error returned below, re-logged by runQueueWorker) carries
	// the user.
	ctx = log.WithLogValues(ctx, "uploadId", t.UploadID, "did", t.RepoDID, "livestream", t.LivestreamURI)
	log.Log(ctx, "dequeued finalize-livestream-vod task")
	if state.livestreamVODFinalizer == nil {
		log.Warn(ctx, "no livestream VOD finalizer configured; dropping task",
			"uploadId", t.UploadID, "did", t.RepoDID)
		return state.CompleteTask(ctx, task.ID)
	}
	if err := state.SetUploadProcessing(ctx, t.UploadID); err != nil {
		log.Warn(ctx, "failed to mark upload as processing", "uploadId", t.UploadID, "error", err)
	}
	cid, err := state.livestreamVODFinalizer(ctx, t)
	if err != nil {
		if ferr := state.SetUploadFailed(ctx, t.UploadID, err.Error()); ferr != nil {
			log.Warn(ctx, "failed to mark upload as failed", "uploadId", t.UploadID, "error", ferr)
		}
		// Flip the tied draft to 'error' so the user sees the failure.
		if derr := state.SetDraftError(ctx, t.UploadID, err.Error()); derr != nil {
			log.Warn(ctx, "failed to mark draft as failed", "uploadId", t.UploadID, "error", derr)
		}
		// Complete so it doesn't retry: most finalize failures are permanent
		// (missing objects, unreadable bytes, no OAuth session).
		_ = state.CompleteTask(ctx, task.ID)
		return fmt.Errorf("finalize livestream VOD upload %s: %w", t.UploadID, err)
	}
	// As with VODProcess, the finalizer calls SetUploadProcessed internally, so
	// re-read the finished Upload row and flip the tied draft to 'ready'.
	if err := state.markDraftReadyFromUpload(ctx, t.UploadID); err != nil {
		log.Warn(ctx, "failed to mark draft ready", "uploadId", t.UploadID, "error", err)
	}
	log.Log(ctx, "livestream VOD finalized", "uploadId", t.UploadID, "cid", cid)
	if t.Publish != nil {
		// The VOD is finalized either way: a failed publish leaves an upload
		// the streamer can still publish from the app, so it is logged, not
		// retried (a retry would finalize and publish the tracks again).
		if state.videoPublisher == nil {
			log.Error(ctx, "finalize-livestream-vod: publish requested but no video publisher configured", "uploadId", t.UploadID)
		} else if uri, vcid, perr := state.videoPublisher(ctx, t); perr != nil {
			log.Error(ctx, "finalize-livestream-vod: VOD finalized but publishing the video record failed", "uploadId", t.UploadID, "error", perr)
		} else {
			log.Log(ctx, "livestream VOD published", "uploadId", t.UploadID, "uri", uri, "cid", vcid)
		}
	}
	return state.CompleteTask(ctx, task.ID)
}

// ViewCountAggregator runs the view-log → place.stream.media.viewCount
// aggregation for one window. Same function-pointer indirection trick
// as VODProcessor: pkg/statedb stays ignorant of pkg/viewlog (which
// pulls in blob storage + atproto publishing).
type ViewCountAggregator func(ctx context.Context, t ViewCountAggregateTask) error

func (state *StatefulDB) SetViewCountAggregator(f ViewCountAggregator) {
	state.viewCountAggregator = f
}

func (state *StatefulDB) processViewCountAggregateTask(ctx context.Context, task *AppTask) error {
	ctx = log.WithLogValues(ctx, "func", "processViewCountAggregateTask")
	var t ViewCountAggregateTask
	if err := json.Unmarshal(task.Payload, &t); err != nil {
		return err
	}
	if state.viewCountAggregator == nil {
		log.Warn(ctx, "no view-count aggregator configured; dropping task",
			"windowStart", t.WindowStart, "windowEnd", t.WindowEnd)
		return state.CompleteTask(ctx, task.ID)
	}
	if err := state.viewCountAggregator(ctx, t); err != nil {
		return fmt.Errorf("view-count aggregation: %w", err)
	}
	return state.CompleteTask(ctx, task.ID)
}

// CDNLogIngester pulls newly archived CDN access logs into the view-log
// store. Same indirection as ViewCountAggregator.
type CDNLogIngester func(ctx context.Context) error

func (state *StatefulDB) SetCDNLogIngester(f CDNLogIngester) {
	state.cdnLogIngester = f
}

func (state *StatefulDB) processCDNLogIngestTask(ctx context.Context, task *AppTask) error {
	ctx = log.WithLogValues(ctx, "func", "processCDNLogIngestTask")
	if state.cdnLogIngester == nil {
		log.Warn(ctx, "no cdn log ingester configured; dropping task")
		return state.CompleteTask(ctx, task.ID)
	}
	if err := state.cdnLogIngester(ctx); err != nil {
		return fmt.Errorf("cdn log ingest: %w", err)
	}
	return state.CompleteTask(ctx, task.ID)
}

func (state *StatefulDB) processFinalizeLivestreamTask(ctx context.Context, task *AppTask) error {
	ctx = log.WithLogValues(ctx, "func", "processFinalizeLivestreamTask")
	log.Debug(ctx, "processing finalize livestream task")
	var finalizeLivestreamTask FinalizeLivestreamTask
	if err := json.Unmarshal(task.Payload, &finalizeLivestreamTask); err != nil {
		return err
	}
	livestream, err := state.model.GetLivestream(finalizeLivestreamTask.LivestreamURI)
	if err != nil {
		return fmt.Errorf("failed to get latest livestream for userDID: %w", err)
	}
	if livestream == nil {
		return fmt.Errorf("no livestream found for URI: %s", finalizeLivestreamTask.LivestreamURI)
	}
	lastLivestreamView, err := livestream.ToLivestreamView()
	if err != nil {
		return fmt.Errorf("failed to convert livestream to streamplace livestream: %w", err)
	}
	rec, ok := lastLivestreamView.Record.Val.(*placestream.Livestream)
	if !ok {
		return fmt.Errorf("livestream is not a streamplace livestream")
	}
	if rec.LastSeenAt == nil {
		return fmt.Errorf("livestream has no last seen at")
	}
	lastSeenTime, err := time.Parse(time.RFC3339, *rec.LastSeenAt)
	if err != nil {
		return fmt.Errorf("could not parse last seen at: %w", err)
	}
	if rec.IdleTimeoutSeconds == nil || *rec.IdleTimeoutSeconds == 0 {
		log.Debug(ctx, "livestream has no idle timeout, skipping finalization", "uri", livestream.URI)
		return nil
	}
	if time.Since(lastSeenTime) < (time.Duration(*rec.IdleTimeoutSeconds) * time.Second) {
		log.Debug(ctx, "livestream is active, skipping finalization", "lastSeenAt", lastSeenTime)
		return nil
	}
	// If this record is still the streamer's latest livestream, do NOT end it
	// on a stale lastSeenAt alone. lastSeenAt only advances via the per-segment
	// heartbeat (StreamSession.doUpdateLivestream), which is coupled to segment
	// arrival and can lag behind actual ingestion — e.g. after an ingest gap
	// long enough to tear down and restart the StreamSession, the new session's
	// heartbeat may not land on this record before the idle timer fires. Ending
	// here would set endedAt on the record the active stream is publishing
	// under, taking the stream pre-live underneath a still-flowing ingest.
	//
	// Instead, reschedule the check for one more idle window: if the stream is
	// truly abandoned the heartbeat stays frozen and we end it on the next
	// pass; if it's a heartbeat-lag artifact, the heartbeat catches up and the
	// rescheduled task hits the "active" early-return above.
	//
	// BUT the heartbeat-lag guard only applies to repos this node is actively
	// ingesting — the heartbeat runs inside StreamSession.doUpdateLivestream,
	// which requires the streamer to have a local OAuth session on this node.
	// If no session exists, there is no heartbeat to wait for and no write
	// access to end the record anyway: the record arrived via firehose sync
	// from an account that never connected here. Rescheduling in that case just
	// respawns the task every idle window forever (the rescheduled key embeds a
	// fresh timestamp, so dedup never fires), flooding the logs and growing the
	// task table without bound. So drop the task instead.
	latest, err := state.model.GetLatestLivestreamForRepo(livestream.RepoDID)
	if err != nil {
		return fmt.Errorf("failed to get latest livestream for repo: %w", err)
	}
	if latest != nil && latest.URI == livestream.URI {
		// Check for a local session before rescheduling. GetSessionByDID
		// returns gorm.ErrRecordNotFound (or nil session via callers that
		// swallow it) when the repo has never logged in here.
		session, err := state.GetSessionByDID(livestream.RepoDID)
		if errors.Is(err, gorm.ErrRecordNotFound) || (err == nil && session == nil) {
			log.Debug(ctx, "stale latest livestream has no local session; dropping finalize task (firehose-observed, no heartbeat to wait for)", "uri", livestream.URI, "lastSeenAt", lastSeenTime)
			return state.CompleteTask(ctx, task.ID)
		}
		if err != nil {
			return fmt.Errorf("failed to get session for finalize-livestream guard: %w", err)
		}
		if heartbeatFrozen(finalizeLivestreamTask, *rec.LastSeenAt) {
			log.Log(ctx, "livestream is latest for repo and its heartbeat has not moved in a full idle window; ending it", "uri", livestream.URI, "lastSeenAt", lastSeenTime)
		} else {
			rescheduledAt := time.Now().Add(time.Duration(*rec.IdleTimeoutSeconds) * time.Second).UTC()
			rescheduledKey := fmt.Sprintf("finalize-livestream::%s::%s", livestream.URI, rescheduledAt.Format(util.ISO8601))
			next := finalizeLivestreamTask
			next.StaleSince = *rec.LastSeenAt
			_, err = state.EnqueueTask(ctx, TaskFinalizeLivestream, next, WithTaskKey(rescheduledKey), WithScheduledAt(rescheduledAt))
			if err != nil {
				return fmt.Errorf("failed to reschedule finalize livestream task: %w", err)
			}
			log.Log(ctx, "livestream is latest for repo but lastSeenAt is stale; rescheduling finalize once to let heartbeat catch up", "uri", livestream.URI, "lastSeenAt", lastSeenTime, "rescheduledAt", rescheduledAt)
			return nil
		}
	}
	return state.EndLivestreamRecord(ctx, livestream, rec)
}

// EndLivestreamRecord sets endedAt (= lastSeenAt) on a livestream record in
// the streamer's repo, with the streamer's stored OAuth session. Used by the
// idle finalize task and by the operator's finalize route for a record the
// streamer never stopped.
func (state *StatefulDB) EndLivestreamRecord(ctx context.Context, livestream *model.Livestream, rec *placestream.Livestream) error {
	session, err := state.GetSessionByDID(livestream.RepoDID)
	if err != nil {
		return fmt.Errorf("failed to get session: %w", err)
	}
	session, err = state.OATProxy.RefreshIfNeeded(session)
	if err != nil {
		return fmt.Errorf("failed to refresh session: %w", err)
	}
	client, err := state.OATProxy.GetXrpcClient(session)
	if err != nil {
		return fmt.Errorf("failed to get xrpc client: %w", err)
	}
	if rec.EndedAt != nil {
		log.Debug(ctx, "livestream has already ended, skipping", "uri", livestream.URI, "endedAt", *rec.EndedAt)
		return nil
	}

	uri, err := syntax.ParseATURI(livestream.URI)
	if err != nil {
		return fmt.Errorf("failed to parse ATURI: %w", err)
	}

	rec.EndedAt = rec.LastSeenAt

	inp := comatproto.RepoPutRecord_Input{
		Collection: "place.stream.livestream",
		Record:     &glex.LexiconTypeDecoder{Val: rec},
		Rkey:       uri.RecordKey().String(),
		Repo:       livestream.RepoDID,
		SwapRecord: &livestream.CID,
	}
	out := comatproto.RepoPutRecord_Output{}

	err = client.Do(ctx, xrpc.Procedure, "application/json", "com.atproto.repo.putRecord", map[string]any{}, inp, &out)
	if err != nil {
		return fmt.Errorf("failed to update livestream record: %w", err)
	}

	log.Log(ctx, "livestream finalized", "uri", livestream.URI, "endedAt", *rec.EndedAt)

	return nil
}

func (state *StatefulDB) processNotificationTask(ctx context.Context, task *AppTask) error {
	var notificationTask NotificationTask
	if err := json.Unmarshal(task.Payload, &notificationTask); err != nil {
		return err
	}
	lsv := notificationTask.Livestream
	rec, err := glex.RecordAs[placestream.Livestream](lsv.Record.Val)
	if err != nil {
		return fmt.Errorf("invalid livestream record: %w", err)
	}
	userDID := lsv.Author.Did

	log.Warn(ctx, "Livestream detected! Blasting followers!", "title", rec.Title, "url", rec.Url, "createdAt", rec.CreatedAt, "repo", userDID)
	followers, err := state.model.GetUserFollowers(ctx, userDID)
	if err != nil {
		return err
	}

	followersDIDs := make([]string, 0, len(followers))
	for _, follower := range followers {
		followersDIDs = append(followersDIDs, follower.UserDID)
	}

	log.Log(ctx, "found followers", "count", len(followersDIDs))

	notifications, err := state.GetManyNotifications(followersDIDs)
	if err != nil {
		return err
	}

	if state.noter != nil {
		nb := &notificationpkg.NotificationBlast{
			Title: fmt.Sprintf("🔴 @%s is LIVE!", lsv.Author.Handle),
			Body:  rec.Title,
			Data: map[string]string{
				"path": fmt.Sprintf("/%s", lsv.Author.Handle),
			},
		}
		targets := make([]notificationpkg.NotificationTarget, len(notifications))
		for i, n := range notifications {
			targets[i] = notificationpkg.NotificationTarget{Token: n.Token, Type: n.Type}
		}
		err = state.noter.Blast(ctx, targets, nb)
		if err != nil {
			log.Error(ctx, "failed to blast notifications", "err", err)
		} else {
			log.Log(ctx, "sent notifications", "user", userDID, "count", len(notifications), "content", nb)
		}
		// Prune web push subscriptions whose endpoints returned 410 Gone /
		// 404 — they're dead and would just fail again on every future blast.
		for _, token := range notificationpkg.ExpiredTokens(err) {
			if delErr := state.DeleteNotification(token); delErr != nil {
				log.Error(ctx, "failed to prune expired notification", "token", token, "err", delErr)
			} else {
				log.Log(ctx, "pruned expired notification", "token", token)
			}
		}
	} else {
		log.Log(ctx, "no notifier configured, skipping notifications", "user", userDID, "count", len(notifications))
	}

	// Send to webhooks using webhook manager
	webhooks, err := state.GetActiveWebhooksForUser(userDID, "livestream")
	if err != nil {
		log.Error(ctx, "failed to get livestream webhooks", "err", err)
	} else {
		for _, w := range webhooks {
			lexiconWebhook, err := w.ToLexicon()
			if err != nil {
				log.Error(ctx, "failed to convert webhook to lexicon", "err", err, "webhook_id", w.ID)
				continue
			}
			go func(lexiconWebhook placestream.ServerDefs_Webhook, wid string) {
				err := webhook.SendLivestreamWebhook(ctx, &lexiconWebhook, notificationTask.PDSURL, &lsv, notificationTask.FeedPost, &notificationTask.ChatProfile)
				if err != nil {
					log.Error(ctx, "failed to send livestream to webhook", "err", err, "webhook_id", wid)
					err := state.IncrementWebhookError(wid)
					if err != nil {
						log.Error(ctx, "failed to increment webhook error count", "err", err, "webhook_id", wid)
					}
				} else {
					log.Log(ctx, "sent livestream to webhook", "webhook_id", wid)
					err := state.ResetWebhookError(wid)
					if err != nil {
						log.Error(ctx, "failed to reset webhook error count", "err", err, "webhook_id", wid)
					}
				}
			}(lexiconWebhook, w.ID)
		}
	}
	return nil
}

func (state *StatefulDB) processStreamReceivedTask(ctx context.Context, task *AppTask) error {
	var streamReceivedTask StreamReceivedTask
	if err := json.Unmarshal(task.Payload, &streamReceivedTask); err != nil {
		return err
	}

	webhooks, err := state.GetActiveWebhooksForUser(streamReceivedTask.StreamerDID, "stream.received")
	if err != nil {
		return fmt.Errorf("failed to get stream.received webhooks: %w", err)
	}
	for _, w := range webhooks {
		lexiconWebhook, err := w.ToLexicon()
		if err != nil {
			log.Error(ctx, "failed to convert webhook to lexicon", "err", err, "webhook_id", w.ID)
			continue
		}
		go func(lexiconWebhook placestream.ServerDefs_Webhook, wid string) {
			err := webhook.SendStreamReceivedWebhook(ctx, &lexiconWebhook, streamReceivedTask.StreamerDID)
			if err != nil {
				log.Error(ctx, "failed to send stream.received webhook", "err", err, "webhook_id", wid)
				err = state.IncrementWebhookError(wid)
				if err != nil {
					log.Error(ctx, "failed to increment webhook error count", "err", err, "webhook_id", wid)
				}
			} else {
				log.Log(ctx, "sent stream.received webhook", "webhook_id", wid)
				err = state.ResetWebhookError(wid)
				if err != nil {
					log.Error(ctx, "failed to reset webhook error count", "err", err, "webhook_id", wid)
				}
			}
		}(lexiconWebhook, w.ID)
	}
	return nil
}

func (state *StatefulDB) processChatMessageTask(ctx context.Context, task *AppTask) error {
	var chatTask ChatTask
	if err := json.Unmarshal(task.Payload, &chatTask); err != nil {
		return err
	}
	scm := chatTask.MessageView
	rec, err := glex.RecordAs[placestream.ChatMessage](scm.Record.Val)
	if err != nil {
		return fmt.Errorf("invalid chat message record: %w", err)
	}

	// Send to webhooks using webhook manager
	webhooks, err := state.GetActiveWebhooksForUser(rec.Streamer, "chat")
	if err != nil {
		log.Error(ctx, "failed to get chat webhooks", "err", err)
	} else {
		for _, w := range webhooks {
			lexiconWebhook, err := w.ToLexicon()
			if err != nil {
				log.Error(ctx, "failed to convert webhook to lexicon", "err", err, "webhook_id", w.ID)
				continue
			}
			go func(lexiconWebhook placestream.ServerDefs_Webhook, wid string) {
				err := webhook.SendChatWebhook(ctx, &lexiconWebhook, scm.Author.Did, &scm)
				if err != nil {
					log.Error(ctx, "failed to send chat to webhook", "err", err, "webhook_id", wid)
					err = state.IncrementWebhookError(wid)
					if err != nil {
						log.Error(ctx, "failed to increment webhook error count", "err", err, "webhook_id", wid)
					}
				} else {
					log.Log(ctx, "sent chat to webhook", "webhook_id", wid)
					err = state.ResetWebhookError(wid)
					if err != nil {
						log.Error(ctx, "failed to reset webhook error count", "err", err, "webhook_id", wid)
					}
				}
			}(lexiconWebhook, w.ID)
		}
	}
	return nil
}