pulse_pixelstream_types/

commands.rs

use myko::prelude::*;
use std::{
    collections::{HashMap, HashSet},
    sync::Arc,
};

use myko::command::{CommandContext, CommandError, CommandHandler};
use myko::entities::client::ClientStatus;
use myko_macros::myko_command;

use crate::cam_pref::{CamPref, CamPrefId};
use crate::camera_home::{CameraHome, CameraHomeId};
use crate::collection::{
    resolve_collection_path, Collection, CollectionId, CollectionQuery, GetCollectionsByIds,
    GetCollectionsByQuery,
};
use crate::collection_membership::{
    CollectionMembership, CollectionMembershipId, CollectionMembershipQuery,
    GetCollectionMembershipsByIds, GetCollectionMembershipsByQuery,
};
use crate::control_lock::{ControlLock, ControlLockId, GetControlLockById};
use crate::frame_capture::FrameCaptureRequestId;
use crate::frame_capture_status::FrameCaptureStatusId;
use crate::frame_capture_target::{
    FrameCaptureTargetSummaryId, GetFrameCaptureTargetSummarysByIds,
};
use crate::legacy_capture_run::{
    CaptureRunQuery as LegacyCaptureRunQuery, GetCaptureRunsByQuery as GetLegacyCaptureRunsByQuery,
};
use crate::legacy_shot_list::{
    GetShotListsByQuery as GetLegacyShotListsByQuery, ShotListQuery as LegacyShotListQuery,
};
use crate::recording_job::{
    CreativeStatus, DeliveryStatus, GetRecordingJobsByIds, GetRecordingJobsByQuery, RecordingJob,
    RecordingJobId, RecordingJobQuery, Take, TakeState,
};
use crate::recording_job_request::{RecordingJobRequest, RecordingJobRequestId};
use crate::recording_job_status::{
    GetRecordingJobStatussByIds, RecordingJobStatus, RecordingJobStatusId,
};
use crate::recording_plan::{RecordingJobAction, RecordingJobPhase, ShotEntryPlan};
use crate::recording_request::{RecordingRequest, RecordingRequestId};
use crate::recording_status::{RecordingState, RecordingStatus, RecordingStatusId};
use crate::shot::{
    effective_library_id, GetShotsByIds, GetShotsByQuery, Shot, ShotDiscovery, ShotId, ShotKind,
    ShotQuery, DEFAULT_SHOT_HOLD_DURATION_MS, DEFAULT_SHOT_LIBRARY_ID,
    DEFAULT_SHOT_ROTATION_SPEED_DEG_S, DEFAULT_SHOT_TRANSLATION_SPEED_CM_S,
};
use crate::stream::{Stream, StreamId};
use crate::timeline::{
    GetTimelinesByIds, GetTimelinesByQuery, ShotDirection, ShotEntry, ShotEntryMode, Timeline,
    TimelineId, TimelineQuery,
};
use crate::viewer::{GetViewerById, GetViewersByQuery, Viewer, ViewerId, ViewerQuery};
use crate::StoredCaptureContext;

const MAX_OTIO_IMPORT_BYTES: usize = 10 * 1024 * 1024;
const MAX_DISCOVERED_SHOTS: usize = 10_000;
const MAX_COLLECTION_DEPTH: usize = 16;
const MAX_COLLECTION_NAME_CHARS: usize = 128;

fn require_uuid_v7_collection_ids() -> bool {
    std::env::var("PULSE_PIXELSTREAM_REQUIRE_UUID_V7_COLLECTION_IDS").is_ok_and(|value| {
        matches!(
            value.trim().to_ascii_lowercase().as_str(),
            "1" | "true" | "yes"
        )
    })
}

fn normalized_timeline_name(name: &str) -> String {
    name.trim().to_lowercase()
}

fn normalized_collection_name(name: &str) -> String {
    name.trim().to_lowercase()
}

/// Project historic `ShotList` rows into OTIO-native Timeline entities.
///
/// Legacy rows remain readable during the compatibility window, but all new
/// commands and queries operate exclusively on Timeline. The migration keeps
/// collection edges and capture history intact while active code uses canonical
/// Timeline ids.
fn migrate_legacy_timelines(
    ctx: &CommandContext,
    shots: &[Shot],
) -> Result<Vec<Timeline>, CommandError> {
    let existing_rows = ctx
        .exec_query(GetTimelinesByQuery(TimelineQuery::default()))?
        .into_iter()
        .map(|timeline| timeline.as_ref().clone())
        .collect::<Vec<_>>();
    let mut timelines_by_id = existing_rows
        .iter()
        .cloned()
        .map(|timeline| (timeline.id.clone(), timeline))
        .collect::<HashMap<_, _>>();
    let mut id_migrations = HashMap::<String, String>::new();
    for legacy in ctx.exec_query(GetLegacyShotListsByQuery(LegacyShotListQuery::default()))? {
        let legacy_id = legacy.id.to_string();
        let mut timeline = legacy.as_ref().clone().into_timeline();
        let timeline_id = timeline.id.to_string();
        if legacy_id != timeline_id {
            id_migrations.insert(legacy_id, timeline_id);
        }
        if let Some(current) = timelines_by_id.get_mut(&timeline.id) {
            // Several old per-stream defaults project to the one shared default.
            // Merge missing source+direction pairs while preserving an already
            // canonical authored sequence and its independently-tuned entries.
            for entry in timeline.entries.drain(..) {
                for direction in entry.directions() {
                    if current.entries.iter().any(|existing| {
                        existing.shot_id == entry.shot_id
                            && existing.directions().contains(direction)
                    }) {
                        continue;
                    }
                    let mut entry = entry.clone();
                    entry.entry_id.clear();
                    entry.direction = ShotEntryMode::from_direction(*direction);
                    current.entries.push(entry);
                }
            }
            current.revision = current.revision.max(timeline.revision);
            current.normalize_entries();
            current.backfill_entry_parameters(shots);
        } else {
            timeline.backfill_entry_parameters(shots);
            timelines_by_id.insert(timeline.id.clone(), timeline);
        }
    }
    for timeline in timelines_by_id.values() {
        let changed = existing_rows
            .iter()
            .find(|existing| existing.id == timeline.id)
            != Some(timeline);
        if changed {
            ctx.emit_set(timeline)?;
        }
    }

    // Canonicalize every organizational edge id, including rows whose Timeline
    // id did not change. Old stable ids embedded the retired entity name.
    for membership in ctx.exec_query(GetCollectionMembershipsByQuery(
        CollectionMembershipQuery::default(),
    ))? {
        let timeline_id = id_migrations
            .get(&membership.timeline_id)
            .cloned()
            .unwrap_or_else(|| membership.timeline_id.clone());
        let replacement_id = CollectionMembershipId::from(CollectionMembership::stable_id(
            &membership.collection_id,
            &timeline_id,
        ));
        if replacement_id == membership.id && timeline_id == membership.timeline_id {
            continue;
        }
        ctx.emit_set(&CollectionMembership {
            id: replacement_id,
            collection_id: membership.collection_id.clone(),
            timeline_id,
            sort_order: membership.sort_order,
        })?;
        ctx.emit_del(membership.as_ref())?;
    }
    for legacy_id in id_migrations.keys() {
        if let Some(old_timeline) = ctx.exec_query_first(GetTimelinesByIds {
            ids: vec![TimelineId::from(legacy_id.clone())],
        })? {
            ctx.emit_del(old_timeline.as_ref())?;
        }
    }
    Ok(timelines_by_id.into_values().collect())
}

fn migrate_legacy_recording_jobs(ctx: &CommandContext) -> Result<(), CommandError> {
    for legacy in ctx.exec_query(GetLegacyCaptureRunsByQuery(LegacyCaptureRunQuery::default()))? {
        let job = legacy.to_recording_job();
        if ctx
            .exec_query_first(GetRecordingJobsByIds {
                ids: vec![job.id.clone()],
            })?
            .is_none()
        {
            ctx.emit_set(&job)?;
        }
    }
    Ok(())
}

/// Idempotent persisted-data migration invoked by OTIO-native clients when
/// they connect. This is deliberately explicit and observable rather than a
/// hidden database rewrite during server startup.
#[myko_command(TimelineId)]
pub struct MigrateOtioTimelines {}

impl CommandHandler for MigrateOtioTimelines {
    fn execute(self, ctx: CommandContext) -> Result<TimelineId, CommandError> {
        let shots = ctx
            .exec_query(GetShotsByQuery(ShotQuery::default()))?
            .into_iter()
            .map(|shot| shot.as_ref().clone())
            .collect::<Vec<_>>();
        let timelines = migrate_legacy_timelines(&ctx, &shots)?;
        apply_timeline_reconciliation(
            &ctx,
            reconcile_timelines(DEFAULT_SHOT_LIBRARY_ID, &shots, timelines),
        )?;
        migrate_legacy_recording_jobs(&ctx)?;
        Ok(TimelineId::from("otio-timeline-migration"))
    }
}
fn timeline_candidate_is_better(candidate: &Timeline, current: &Timeline) -> bool {
    let candidate_rank = (
        u8::from(!candidate.library_id.trim().is_empty()),
        u8::from(candidate.streamer_id.trim().is_empty()),
    );
    let current_rank = (
        u8::from(!current.library_id.trim().is_empty()),
        u8::from(current.streamer_id.trim().is_empty()),
    );
    candidate_rank > current_rank
        || (candidate_rank == current_rank && candidate.id.to_string() < current.id.to_string())
}

fn preferred_shot_ids(library_id: &str, shots: &[Shot]) -> HashMap<String, String> {
    let mut preferred = HashMap::<(u8, String), &Shot>::new();
    for candidate in shots
        .iter()
        .filter(|shot| shot.effective_library_id() == library_id)
    {
        let key = (
            shot_kind_key(&candidate.kind),
            candidate.target_name.clone(),
        );
        let replace = preferred.get(&key).is_none_or(|current| {
            let candidate_rank = u8::from(!candidate.library_id.trim().is_empty());
            let current_rank = u8::from(!current.library_id.trim().is_empty());
            candidate_rank > current_rank
                || (candidate_rank == current_rank
                    && candidate.id.to_string() < current.id.to_string())
        });
        if replace {
            preferred.insert(key, candidate);
        }
    }

    shots
        .iter()
        .filter(|shot| shot.effective_library_id() == library_id)
        .filter_map(|shot| {
            preferred
                .get(&(shot_kind_key(&shot.kind), shot.target_name.clone()))
                .map(|winner| (shot.id.to_string(), winner.id.to_string()))
        })
        .collect()
}

#[derive(Default)]
struct TimelineReconciliation {
    upserts: Vec<Timeline>,
    deletes: Vec<Timeline>,
    id_migrations: HashMap<String, String>,
}

fn apply_timeline_reconciliation(
    ctx: &CommandContext,
    reconciliation: TimelineReconciliation,
) -> Result<(), CommandError> {
    for timeline in &reconciliation.upserts {
        ctx.emit_set(timeline)?;
    }

    if !reconciliation.id_migrations.is_empty() {
        let memberships = ctx
            .exec_query(GetCollectionMembershipsByQuery(
                CollectionMembershipQuery::default(),
            ))?
            .into_iter()
            .map(|membership| membership.as_ref().clone())
            .collect::<Vec<_>>();
        for membership in &memberships {
            let Some(timeline_id) = reconciliation.id_migrations.get(&membership.timeline_id)
            else {
                continue;
            };
            let replacement_id = CollectionMembershipId::from(CollectionMembership::stable_id(
                &membership.collection_id,
                timeline_id,
            ));
            let sort_order = memberships
                .iter()
                .filter(|candidate| candidate.id == replacement_id)
                .map(|candidate| candidate.sort_order)
                .chain(std::iter::once(membership.sort_order))
                .min()
                .unwrap_or(membership.sort_order);
            let replaces_membership = replacement_id != membership.id;
            ctx.emit_set(&CollectionMembership {
                id: replacement_id,
                collection_id: membership.collection_id.clone(),
                timeline_id: timeline_id.clone(),
                sort_order,
            })?;
            if replaces_membership {
                ctx.emit_del(membership)?;
            }
        }
    }

    for timeline in &reconciliation.deletes {
        ctx.emit_del(timeline)?;
    }
    Ok(())
}

/// Collapse legacy per-stream timelines into one persistent shared definition per
/// case-insensitive name. Membership is merged rather than discarded, and entry
/// references follow the same preferred Shot projection used by the UI.
fn reconcile_timelines(
    library_id: &str,
    shots: &[Shot],
    timelines: Vec<Timeline>,
) -> TimelineReconciliation {
    let library_id = effective_library_id(library_id);
    let preferred_shots = preferred_shot_ids(library_id, shots);
    let mut groups = HashMap::<String, Vec<Timeline>>::new();
    for timeline in timelines
        .into_iter()
        .filter(|timeline| timeline.effective_library_id() == library_id)
    {
        let name = if timeline.has_legacy_default_identity() && timeline.has_legacy_default_name() {
            "\0legacy-default-timeline".to_owned()
        } else {
            normalized_timeline_name(&timeline.name)
        };
        if !name.is_empty() {
            groups.entry(name).or_default().push(timeline);
        }
    }

    let mut reconciliation = TimelineReconciliation::default();
    for (_, mut group) in groups {
        group.sort_by_key(|timeline| timeline.id.to_string());
        let preferred = group
            .iter()
            .reduce(|current, candidate| {
                if timeline_candidate_is_better(candidate, current) {
                    candidate
                } else {
                    current
                }
            })
            .expect("timeline groups are non-empty");
        let is_legacy_default = group.iter().any(|timeline| {
            timeline.has_legacy_default_identity() && timeline.has_legacy_default_name()
        });
        let canonical_id = if is_legacy_default {
            TimelineId::from(Timeline::legacy_default_id(library_id))
        } else {
            preferred.id.clone()
        };

        // The preferred timeline is the authoritative OTIO-like sequence and may
        // intentionally contain the same Shot more than once. Secondary legacy
        // rows contribute only missing shot+direction pairs, preventing old
        // per-stream copies from multiplying an already-authored sequence.
        let mut entries = preferred
            .entries
            .iter()
            .cloned()
            .map(|mut entry| {
                entry.shot_id = preferred_shots
                    .get(&entry.shot_id)
                    .cloned()
                    .unwrap_or(entry.shot_id);
                entry
            })
            .collect::<Vec<_>>();
        for timeline in group.iter().filter(|timeline| timeline.id != preferred.id) {
            for source in &timeline.entries {
                let shot_id = preferred_shots
                    .get(&source.shot_id)
                    .cloned()
                    .unwrap_or_else(|| source.shot_id.clone());
                for direction in source.directions() {
                    let already_present = entries.iter().any(|entry| {
                        entry.shot_id == shot_id && entry.directions().contains(direction)
                    });
                    if !already_present {
                        let mut entry = source.clone();
                        entry.entry_id.clear();
                        entry.shot_id = shot_id.clone();
                        entry.direction = ShotEntryMode::from_direction(*direction);
                        entries.push(entry);
                    }
                }
            }
        }
        let revision = group
            .iter()
            .map(|timeline| timeline.revision)
            .max()
            .unwrap_or(0);
        let mut canonical = Timeline {
            id: canonical_id.clone(),
            library_id: library_id.to_owned(),
            streamer_id: String::new(),
            name: if is_legacy_default && preferred.has_legacy_default_name() {
                "Migrated timeline".to_owned()
            } else {
                preferred.name.trim().to_owned()
            },
            revision,
            entries,
            sort_order: group
                .iter()
                .map(|timeline| timeline.sort_order)
                .min()
                .unwrap_or(preferred.sort_order),
        };
        canonical.normalize_entries();
        canonical.backfill_entry_parameters(shots);

        for timeline in &group {
            if timeline.id != canonical_id {
                reconciliation
                    .id_migrations
                    .insert(timeline.id.to_string(), canonical_id.to_string());
            }
        }

        if group.iter().find(|timeline| timeline.id == canonical_id) != Some(&canonical) {
            reconciliation.upserts.push(canonical);
        }
        reconciliation.deletes.extend(
            group
                .into_iter()
                .filter(|timeline| timeline.id != canonical_id),
        );
    }
    reconciliation
}

fn ensure_unique_timeline_name(
    ctx: &CommandContext,
    library_id: &str,
    id: &TimelineId,
    name: &str,
) -> Result<(), CommandError> {
    let normalized_name = normalized_timeline_name(name);
    if normalized_name.is_empty() {
        return Err(command_error(ctx, "Timeline name cannot be empty"));
    }
    let duplicate = ctx
        .exec_query(GetTimelinesByQuery(TimelineQuery::default()))?
        .into_iter()
        .any(|timeline| {
            timeline.id != *id
                && timeline.effective_library_id() == effective_library_id(library_id)
                && normalized_timeline_name(&timeline.name) == normalized_name
        });
    if duplicate {
        return Err(command_error(
            ctx,
            format!("A timeline named ‘{}’ already exists", name.trim()),
        ));
    }
    Ok(())
}

fn shot_kind_key(kind: &ShotKind) -> u8 {
    match kind {
        ShotKind::Moving => 0,
        ShotKind::Static => 1,
    }
}

fn discovered_shots_to_create(
    library_id: &str,
    discoveries: Vec<ShotDiscovery>,
    existing: &[Shot],
) -> Vec<Shot> {
    let library_id = effective_library_id(library_id);
    let mut known = existing
        .iter()
        .filter(|shot| shot.effective_library_id() == library_id)
        .map(|shot| (shot_kind_key(&shot.kind), shot.target_name.clone()))
        .collect::<HashSet<_>>();
    let mut next_index = existing
        .iter()
        .filter(|shot| shot.effective_library_id() == library_id)
        .map(|shot| shot.shot_index)
        .max()
        .map_or(0, |index| index.saturating_add(1));

    discoveries
        .into_iter()
        .filter_map(|discovery| {
            let name = discovery.name.trim();
            let target_name = discovery.target_name.trim();
            if name.is_empty() || target_name.is_empty() {
                return None;
            }
            let key = (shot_kind_key(&discovery.kind), target_name.to_owned());
            if !known.insert(key) {
                return None;
            }
            let shot = Shot {
                id: ShotId::from(Shot::stable_id(library_id, &discovery.kind, target_name)),
                library_id: library_id.to_owned(),
                streamer_id: String::new(),
                name: name.to_owned(),
                kind: discovery.kind,
                target_name: target_name.to_owned(),
                translation_speed_cm_s: DEFAULT_SHOT_TRANSLATION_SPEED_CM_S,
                rotation_speed_deg_s: DEFAULT_SHOT_ROTATION_SPEED_DEG_S,
                hold_duration_ms: DEFAULT_SHOT_HOLD_DURATION_MS,
                travel_duration_ms: crate::DEFAULT_SHOT_TRAVEL_DURATION_MS,
                default_entry_mode: ShotEntryMode::Forward,
                shot_index: next_index,
            };
            next_index = next_index.saturating_add(1);
            Some(shot)
        })
        .collect()
}

/// Reconcile live camera targets into a persistent shot library. This command
/// is create-only by `(library, kind, target_name)`: discovery can add a new
/// rail or preset, but can never overwrite an operator's tuned Shot values.
#[myko_command]
pub struct DiscoverShots {
    #[serde(default)]
    pub library_id: String,
    pub shots: Vec<ShotDiscovery>,
}

impl CommandHandler for DiscoverShots {
    fn execute(self, ctx: CommandContext) -> Result<(), CommandError> {
        if self.shots.len() > MAX_DISCOVERED_SHOTS {
            return Err(command_error(
                &ctx,
                format!("shot discovery exceeds the {MAX_DISCOVERED_SHOTS} target limit"),
            ));
        }
        let mut existing = ctx
            .exec_query(GetShotsByQuery(ShotQuery::default()))?
            .into_iter()
            .map(|shot| shot.as_ref().clone())
            .collect::<Vec<_>>();
        let created = discovered_shots_to_create(&self.library_id, self.shots, &existing);
        for shot in &created {
            ctx.emit_set(shot)?;
        }
        existing.extend(created);

        let timelines = ctx
            .exec_query(GetTimelinesByQuery(TimelineQuery::default()))?
            .into_iter()
            .map(|timeline| timeline.as_ref().clone())
            .collect();
        let reconciliation = reconcile_timelines(&self.library_id, &existing, timelines);
        apply_timeline_reconciliation(&ctx, reconciliation)
    }
}

/// Upsert one persistent shot definition. The stable shot ID is supplied by the
/// client so reconnects cannot create duplicate definitions for the same target.
#[myko_command(ShotId)]
pub struct SetShot {
    pub shot_id: ShotId,
    #[serde(default)]
    pub library_id: String,
    #[serde(default)]
    pub streamer_id: String,
    pub name: String,
    #[serde(alias = "target_kind")]
    pub kind: ShotKind,
    pub target_name: String,
    pub translation_speed_cm_s: f32,
    pub rotation_speed_deg_s: f32,
    pub hold_duration_ms: u64,
    #[serde(default = "crate::default_shot_travel_duration_ms")]
    pub travel_duration_ms: u64,
    #[serde(
        default,
        alias = "defaultClipMode",
        alias = "defaultListMode",
        alias = "enabled",
        alias = "batchMode"
    )]
    pub default_entry_mode: ShotEntryMode,
    /// Persistent shot label. Keep the established wire name so currently
    /// deployed clients remain compatible while Rust uses the correct taxonomy.
    #[serde(rename = "sortOrder", alias = "shotIndex")]
    pub shot_index: u32,
}

impl CommandHandler for SetShot {
    fn execute(self, ctx: CommandContext) -> Result<ShotId, CommandError> {
        let id = self.shot_id;
        ctx.emit_set(&Shot {
            id: id.clone(),
            library_id: effective_library_id(&self.library_id).to_owned(),
            streamer_id: self.streamer_id,
            name: self.name,
            kind: self.kind,
            target_name: self.target_name,
            translation_speed_cm_s: self.translation_speed_cm_s,
            rotation_speed_deg_s: self.rotation_speed_deg_s,
            hold_duration_ms: self.hold_duration_ms,
            travel_duration_ms: self.travel_duration_ms,
            default_entry_mode: self.default_entry_mode,
            shot_index: self.shot_index,
        })?;
        Ok(id)
    }
}

/// Create or update one named reusable timeline. Every entry persists a complete
/// parameter snapshot; Shot values are defaults used only when adding or
/// migrating an entry.
#[myko_command(TimelineId)]
pub struct SetTimeline {
    #[serde(alias = "shotListId")]
    pub timeline_id: TimelineId,
    #[serde(default)]
    pub library_id: String,
    #[serde(default)]
    pub streamer_id: String,
    pub name: String,
    #[serde(alias = "clips", alias = "cues")]
    pub entries: Vec<ShotEntry>,
    pub sort_order: u32,
}

impl CommandHandler for SetTimeline {
    fn execute(self, ctx: CommandContext) -> Result<TimelineId, CommandError> {
        let id = self.timeline_id;
        ensure_unique_timeline_name(&ctx, &self.library_id, &id, &self.name)?;
        let current = ctx.exec_query_first(GetTimelinesByIds {
            ids: vec![id.clone()],
        })?;
        let revision = current.map(|current| current.next_revision()).unwrap_or(1);
        let mut timeline = Timeline {
            id: id.clone(),
            library_id: effective_library_id(&self.library_id).to_owned(),
            streamer_id: self.streamer_id,
            name: self.name.trim().to_owned(),
            revision,
            entries: self.entries,
            sort_order: self.sort_order,
        };
        timeline.normalize_entries();
        let shots = ctx
            .exec_query(GetShotsByQuery(ShotQuery::default()))?
            .into_iter()
            .map(|shot| shot.as_ref().clone())
            .collect::<Vec<_>>();
        timeline.backfill_entry_parameters(&shots);
        ctx.emit_set(&timeline)?;
        Ok(id)
    }
}

// ─────────────────────────────────────────────────────────────────────────
// Entry-addressed timeline editing
//
// `SetTimeline` replaces the whole `entries` vector, so two operators editing
// one timeline silently overwrite each other — and timeline editing is not
// wheel-gated, so that is an ordinary Tuesday, not a race you have to try for.
// Adding an expected-revision check would only turn the lost update into a
// failed save. These commands name the entry they act on instead, so
// concurrent edits to different entries compose and nobody's work disappears.
// ─────────────────────────────────────────────────────────────────────────

/// Load a timeline for editing, or explain why it cannot be edited.
fn timeline_for_edit(
    ctx: &CommandContext,
    timeline_id: &TimelineId,
) -> Result<Timeline, CommandError> {
    ctx.exec_query_first(GetTimelinesByIds {
        ids: vec![timeline_id.clone()],
    })?
    .map(|timeline| timeline.as_ref().clone())
    .ok_or_else(|| command_error(ctx, "That timeline no longer exists"))
}

/// Persist an edited timeline, advancing the program revision.
fn commit_timeline(ctx: &CommandContext, mut timeline: Timeline) -> Result<(), CommandError> {
    timeline.revision = timeline.next_revision();
    timeline.normalize_entries();
    let shots = ctx
        .exec_query(GetShotsByQuery(ShotQuery::default()))?
        .into_iter()
        .map(|shot| shot.as_ref().clone())
        .collect::<Vec<_>>();
    timeline.backfill_entry_parameters(&shots);
    ctx.emit_set(&timeline)?;
    Ok(())
}

/// Add one shot to a timeline, optionally at a position rather than the end.
#[myko_command(TimelineId)]
pub struct AddShotEntry {
    pub timeline_id: TimelineId,
    pub shot_id: String,
    #[serde(default)]
    pub direction: ShotDirection,
    /// Client-minted so a retried submit cannot double-add.
    pub entry_id: String,
    /// Zero-based; `None` appends.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub position: Option<u32>,
}

impl CommandHandler for AddShotEntry {
    fn execute(self, ctx: CommandContext) -> Result<TimelineId, CommandError> {
        let mut timeline = timeline_for_edit(&ctx, &self.timeline_id)?;
        // Idempotent: the same entry id twice is one entry.
        if timeline.entry_position(&self.entry_id).is_some() {
            return Ok(self.timeline_id);
        }
        let shot = ctx
            .exec_query_first(GetShotsByIds {
                ids: vec![ShotId::from(self.shot_id.clone())],
            })?
            .ok_or_else(|| command_error(&ctx, "That shot no longer exists"))?;
        timeline.insert_shot_entry(
            shot.as_ref(),
            self.direction,
            self.entry_id,
            self.position.map(|position| position as usize),
        );
        commit_timeline(&ctx, timeline)?;
        Ok(self.timeline_id)
    }
}

/// Remove one entry. Naming the entry means a concurrent add elsewhere in the
/// timeline survives.
#[myko_command(TimelineId)]
pub struct RemoveShotEntry {
    pub timeline_id: TimelineId,
    pub entry_id: String,
}

impl CommandHandler for RemoveShotEntry {
    fn execute(self, ctx: CommandContext) -> Result<TimelineId, CommandError> {
        let mut timeline = timeline_for_edit(&ctx, &self.timeline_id)?;
        if timeline.entry_position(&self.entry_id).is_none() {
            // Already gone: someone else removed it, which is the outcome asked for.
            return Ok(self.timeline_id);
        }
        timeline.remove_entry(&self.entry_id);
        commit_timeline(&ctx, timeline)?;
        Ok(self.timeline_id)
    }
}

/// Move one entry to a position in the program.
#[myko_command(TimelineId)]
pub struct MoveShotEntry {
    pub timeline_id: TimelineId,
    pub entry_id: String,
    /// Zero-based target position; clamped to the ends.
    pub position: u32,
}

impl CommandHandler for MoveShotEntry {
    fn execute(self, ctx: CommandContext) -> Result<TimelineId, CommandError> {
        let mut timeline = timeline_for_edit(&ctx, &self.timeline_id)?;
        let Some(current) = timeline.entry_position(&self.entry_id) else {
            return Err(command_error(
                &ctx,
                "That shot is no longer in this timeline",
            ));
        };
        let target = (self.position as usize).min(timeline.entries.len().saturating_sub(1));
        if current == target {
            return Ok(self.timeline_id);
        }
        timeline.move_entry(&self.entry_id, target);
        commit_timeline(&ctx, timeline)?;
        Ok(self.timeline_id)
    }
}

/// Set one entry's capture direction.
#[myko_command(TimelineId)]
pub struct SetShotEntryDirection {
    pub timeline_id: TimelineId,
    pub entry_id: String,
    pub direction: ShotDirection,
}

impl CommandHandler for SetShotEntryDirection {
    fn execute(self, ctx: CommandContext) -> Result<TimelineId, CommandError> {
        let mut timeline = timeline_for_edit(&ctx, &self.timeline_id)?;
        if timeline.entry_position(&self.entry_id).is_none() {
            return Err(command_error(
                &ctx,
                "That shot is no longer in this timeline",
            ));
        }
        timeline.set_entry_direction(&self.entry_id, self.direction);
        commit_timeline(&ctx, timeline)?;
        Ok(self.timeline_id)
    }
}

/// Delete one timeline and all of its organizational placements without
/// deleting any reusable shot definitions.
#[myko_command(TimelineId)]
pub struct RemoveTimeline {
    #[serde(alias = "shotListId")]
    pub timeline_id: TimelineId,
}

impl CommandHandler for RemoveTimeline {
    fn execute(self, ctx: CommandContext) -> Result<TimelineId, CommandError> {
        let id = self.timeline_id;
        let current = ctx
            .exec_query_first(GetTimelinesByIds {
                ids: vec![id.clone()],
            })?
            .ok_or_else(|| command_error(&ctx, format!("Timeline {id} does not exist")))?;
        for membership in ctx
            .exec_query(GetCollectionMembershipsByQuery(
                CollectionMembershipQuery::default(),
            ))?
            .into_iter()
            .filter(|membership| membership.timeline_id == id.to_string())
        {
            ctx.emit_del(membership.as_ref())?;
        }
        ctx.emit_del(current.as_ref())?;
        Ok(id)
    }
}

/// Create or update a generic organizational bin. Collection names and depth
/// are user-defined; the server enforces only tree integrity and sibling-name
/// uniqueness.
#[myko_command(CollectionId)]
pub struct SetCollection {
    pub collection_id: CollectionId,
    #[serde(default)]
    pub library_id: String,
    #[serde(default)]
    pub parent_id: String,
    pub name: String,
    #[serde(default)]
    pub sort_order: u32,
    #[serde(default)]
    pub metadata: HashMap<String, String>,
}

impl CommandHandler for SetCollection {
    fn execute(self, ctx: CommandContext) -> Result<CollectionId, CommandError> {
        let id = self.collection_id;
        let library_id = effective_library_id(&self.library_id).to_owned();
        let name = self.name.trim().to_owned();
        if name.is_empty() {
            return Err(command_error(&ctx, "Collection name cannot be empty"));
        }
        if name.chars().count() > MAX_COLLECTION_NAME_CHARS {
            return Err(command_error(
                &ctx,
                format!("Collection names are limited to {MAX_COLLECTION_NAME_CHARS} characters"),
            ));
        }
        let mut collections = ctx
            .exec_query(GetCollectionsByQuery(CollectionQuery::default()))?
            .into_iter()
            .map(|collection| collection.as_ref().clone())
            .collect::<Vec<_>>();
        if !collections.iter().any(|collection| collection.id == id)
            && !crate::collection::is_uuid_v7(id.as_ref())
        {
            if require_uuid_v7_collection_ids() {
                return Err(command_error(
                    &ctx,
                    "New collection IDs must be UUIDv7; existing legacy collections remain editable",
                ));
            }
            eprintln!(
                "accepted legacy native collection id during UUIDv7 rollout: {}",
                id.as_ref()
            );
        }
        if collections.iter().any(|collection| {
            collection.id != id
                && collection.library_id == library_id
                && collection.parent_id == self.parent_id
                && normalized_collection_name(&collection.name) == normalized_collection_name(&name)
        }) {
            return Err(command_error(
                &ctx,
                format!("A collection named ‘{name}’ already exists here"),
            ));
        }
        if !self.parent_id.trim().is_empty() {
            let parent = collections
                .iter()
                .find(|collection| collection.id.to_string() == self.parent_id)
                .ok_or_else(|| command_error(&ctx, "Parent collection does not exist"))?;
            if parent.library_id != library_id {
                return Err(command_error(
                    &ctx,
                    "A collection cannot be moved between libraries",
                ));
            }
            if parent.id == id {
                return Err(command_error(&ctx, "A collection cannot contain itself"));
            }
            let parent_path = resolve_collection_path(&self.parent_id, &collections)
                .map_err(|error| command_error(&ctx, error))?;
            if parent_path.len() >= MAX_COLLECTION_DEPTH {
                return Err(command_error(
                    &ctx,
                    format!("Collections are limited to {MAX_COLLECTION_DEPTH} levels"),
                ));
            }
            if parent_path
                .iter()
                .any(|segment| segment.collection_id == id.to_string())
            {
                return Err(command_error(
                    &ctx,
                    "A collection cannot be moved inside one of its descendants",
                ));
            }
        }
        let collection = Collection {
            id: id.clone(),
            library_id,
            parent_id: self.parent_id,
            name,
            sort_order: self.sort_order,
            metadata: self.metadata,
        };
        if let Some(current) = collections.iter_mut().find(|row| row.id == id) {
            *current = collection.clone();
        } else {
            collections.push(collection.clone());
        }
        // Validate the resulting row too, including pre-existing corrupt paths.
        resolve_collection_path(id.as_ref(), &collections)
            .map_err(|error| command_error(&ctx, error))?;
        ctx.emit_set(&collection)?;
        Ok(id)
    }
}

/// Remove an empty collection. Memberships directly inside it are removed,
/// while child collections must be deliberately handled first so a broad tree
/// cannot disappear from one accidental click.
#[myko_command(CollectionId)]
pub struct RemoveCollection {
    pub collection_id: CollectionId,
}

impl CommandHandler for RemoveCollection {
    fn execute(self, ctx: CommandContext) -> Result<CollectionId, CommandError> {
        let id = self.collection_id;
        let current = ctx
            .exec_query_first(GetCollectionsByIds {
                ids: vec![id.clone()],
            })?
            .ok_or_else(|| command_error(&ctx, format!("Collection {id} does not exist")))?;
        let has_children = ctx
            .exec_query(GetCollectionsByQuery(CollectionQuery::default()))?
            .into_iter()
            .any(|collection| collection.parent_id == id.to_string());
        if has_children {
            return Err(command_error(
                &ctx,
                "Move or remove child collections before deleting this collection",
            ));
        }
        for membership in ctx
            .exec_query(GetCollectionMembershipsByQuery(
                CollectionMembershipQuery::default(),
            ))?
            .into_iter()
            .filter(|membership| membership.collection_id == id.to_string())
        {
            ctx.emit_del(membership.as_ref())?;
        }
        ctx.emit_del(current.as_ref())?;
        Ok(id)
    }
}

/// Add or remove one many-to-many timeline placement.
#[myko_command(CollectionMembershipId)]
pub struct SetCollectionMembership {
    pub collection_id: String,
    #[serde(alias = "shotListId")]
    pub timeline_id: String,
    pub included: bool,
    #[serde(default)]
    pub sort_order: u32,
}

impl CommandHandler for SetCollectionMembership {
    fn execute(self, ctx: CommandContext) -> Result<CollectionMembershipId, CommandError> {
        let id = CollectionMembershipId::from(CollectionMembership::stable_id(
            &self.collection_id,
            &self.timeline_id,
        ));
        let existing = ctx.exec_query_first(GetCollectionMembershipsByIds {
            ids: vec![id.clone()],
        })?;
        if !self.included {
            if let Some(existing) = existing {
                ctx.emit_del(existing.as_ref())?;
            }
            return Ok(id);
        }
        let collection = ctx
            .exec_query_first(GetCollectionsByIds {
                ids: vec![CollectionId::from(self.collection_id.clone())],
            })?
            .ok_or_else(|| command_error(&ctx, "Collection does not exist"))?;
        let timeline = ctx
            .exec_query_first(GetTimelinesByIds {
                ids: vec![TimelineId::from(self.timeline_id.clone())],
            })?
            .ok_or_else(|| command_error(&ctx, "Timeline does not exist"))?;
        if collection.library_id != timeline.effective_library_id() {
            return Err(command_error(
                &ctx,
                "Collection and timeline belong to different libraries",
            ));
        }
        ctx.emit_set(&CollectionMembership {
            id: id.clone(),
            collection_id: self.collection_id,
            timeline_id: self.timeline_id,
            sort_order: self.sort_order,
        })?;
        Ok(id)
    }
}

/// Atomically import a Pulse-profile OTIO document into the shared definition
/// library. Camera execution remains untouched; this command only persists Shot
/// and Timeline definitions.
#[myko_command(TimelineId)]
pub struct ImportTimelineOtio {
    pub otio_json: String,
    pub sort_order: u32,
}

impl CommandHandler for ImportTimelineOtio {
    fn execute(self, ctx: CommandContext) -> Result<TimelineId, CommandError> {
        if self.otio_json.len() > MAX_OTIO_IMPORT_BYTES {
            return Err(command_error(
                &ctx,
                format!(
                    "OTIO import exceeds the {} byte limit",
                    MAX_OTIO_IMPORT_BYTES
                ),
            ));
        }
        let imported = crate::import_timeline_otio_json(&self.otio_json)
            .map_err(|error| command_error(&ctx, error.to_string()))?;
        if imported.shots.len() > 10_000 {
            return Err(command_error(&ctx, "OTIO import contains too many shots"));
        }
        ensure_unique_timeline_name(
            &ctx,
            imported.timeline.effective_library_id(),
            &imported.timeline.id,
            &imported.timeline.name,
        )?;

        let mut timeline = imported.timeline;
        timeline.backfill_entry_parameters(&imported.shots);
        for shot in imported.shots {
            ctx.emit_set(&shot)?;
        }

        if let Some(current) = ctx.exec_query_first(GetTimelinesByIds {
            ids: vec![timeline.id.clone()],
        })? {
            timeline.revision = timeline.revision.max(current.next_revision());
        }
        timeline.sort_order = self.sort_order;
        timeline.normalize_entries();
        let id = timeline.id.clone();
        ctx.emit_set(&timeline)?;
        Ok(id)
    }
}

/// Atomically import a generic OTIO `SerializableCollection` subtree, including
/// nested bins, reusable timeline placements, timelines, and shot definitions.
#[myko_command(CollectionId)]
pub struct ImportCollectionOtio {
    pub otio_json: String,
}

impl CommandHandler for ImportCollectionOtio {
    fn execute(self, ctx: CommandContext) -> Result<CollectionId, CommandError> {
        if self.otio_json.len() > MAX_OTIO_IMPORT_BYTES {
            return Err(command_error(
                &ctx,
                format!(
                    "OTIO import exceeds the {} byte limit",
                    MAX_OTIO_IMPORT_BYTES
                ),
            ));
        }
        let mut imported = crate::import_collection_otio_json(&self.otio_json)
            .map_err(|error| command_error(&ctx, error.to_string()))?;
        if imported.collections.is_empty() {
            return Err(command_error(&ctx, "OTIO collection is empty"));
        }
        if imported.collections.len() > 10_000
            || imported.memberships.len() > 100_000
            || imported.timelines.len() > 10_000
            || imported.shots.len() > 10_000
        {
            return Err(command_error(&ctx, "OTIO collection exceeds import limits"));
        }
        let root_id = imported.collections[0].id.clone();
        let imported_collection_ids = imported
            .collections
            .iter()
            .map(|collection| collection.id.to_string())
            .collect::<HashSet<_>>();
        let existing_collections = ctx
            .exec_query(GetCollectionsByQuery(CollectionQuery::default()))?
            .into_iter()
            .map(|collection| collection.as_ref().clone())
            .collect::<Vec<_>>();
        for collection in &imported.collections {
            let duplicate = imported
                .collections
                .iter()
                .chain(
                    existing_collections
                        .iter()
                        .filter(|existing| !imported_collection_ids.contains(existing.id.as_ref())),
                )
                .any(|candidate| {
                    candidate.id != collection.id
                        && candidate.library_id == collection.library_id
                        && candidate.parent_id == collection.parent_id
                        && normalized_collection_name(&candidate.name)
                            == normalized_collection_name(&collection.name)
                });
            if duplicate {
                return Err(command_error(
                    &ctx,
                    format!(
                        "A collection named ‘{}’ already exists at the imported location",
                        collection.name
                    ),
                ));
            }
        }

        let imported_timeline_ids = imported
            .timelines
            .iter()
            .map(|timeline| timeline.id.to_string())
            .collect::<HashSet<_>>();
        let existing_timelines = ctx
            .exec_query(GetTimelinesByQuery(TimelineQuery::default()))?
            .into_iter()
            .map(|timeline| timeline.as_ref().clone())
            .collect::<Vec<_>>();
        for timeline in &imported.timelines {
            let duplicate = imported
                .timelines
                .iter()
                .chain(
                    existing_timelines
                        .iter()
                        .filter(|existing| !imported_timeline_ids.contains(existing.id.as_ref())),
                )
                .any(|candidate| {
                    candidate.id != timeline.id
                        && candidate.effective_library_id() == timeline.effective_library_id()
                        && normalized_timeline_name(&candidate.name)
                            == normalized_timeline_name(&timeline.name)
                });
            if duplicate {
                return Err(command_error(
                    &ctx,
                    format!("A timeline named ‘{}’ already exists", timeline.name),
                ));
            }
        }

        for shot in imported.shots {
            ctx.emit_set(&shot)?;
        }
        for timeline in &mut imported.timelines {
            timeline.normalize_entries();
            if let Some(current) = existing_timelines
                .iter()
                .find(|current| current.id == timeline.id)
            {
                timeline.revision = timeline.revision.max(current.next_revision());
            }
            ctx.emit_set(timeline)?;
        }
        for collection in imported.collections {
            ctx.emit_set(&collection)?;
        }
        for membership in imported.memberships {
            ctx.emit_set(&membership)?;
        }
        Ok(root_id)
    }
}

fn command_error(ctx: &CommandContext, message: impl Into<String>) -> CommandError {
    CommandError {
        tx: ctx.tx().to_string(),
        command_id: ctx.command_id.to_string(),
        message: message.into(),
    }
}

/// Create, resume, or cancel a durable recorder-controller-owned RecordingJob.
#[myko_command(RecordingJobRequestId)]
pub struct ControlRecordingJob {
    pub streamer_id: String,
    #[serde(alias = "runId")]
    pub job_id: String,
    pub command_id: String,
    pub action: RecordingJobAction,
    /// Required for Start and frozen for the lifetime of the job.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    #[ts(type = "unknown")]
    pub capture_context: Option<crate::CaptureContext>,
    /// Required for Start. The server resolves the authoritative Timeline and
    /// snapshots its current revision; clients never author execution provenance.
    #[serde(default, alias = "shotListId")]
    pub timeline_id: String,
    /// Optional organizational placement selected by the client. Start
    /// validates the membership and snapshots its root-to-leaf path.
    #[serde(default)]
    pub collection_id: String,
    /// Legacy client-resolved plan. Start ignores this and resolves the
    /// authoritative persisted Timeline; resume/cancel do not need a plan.
    #[serde(default, alias = "shots", alias = "items")]
    pub entries: Vec<ShotEntryPlan>,
    /// Start only. When non-empty, record just these timeline entry ids (a
    /// single shot, or a subset) instead of the whole timeline. Empty = the
    /// whole timeline. The entries still come from the authoritative persisted
    /// Timeline — this only narrows which of its entries run — so take
    /// numbering, delivery placement, and integrity are unchanged.
    #[serde(default)]
    pub entry_ids: Vec<String>,
    #[serde(default)]
    pub preset_duration_ms: u64,
    #[serde(default)]
    pub translation_speed_cm_s: f32,
    #[serde(default)]
    pub rotation_speed_deg_s: f32,
    #[serde(default)]
    pub requested_at_ms: u64,
}

impl CommandHandler for ControlRecordingJob {
    fn execute(self, ctx: CommandContext) -> Result<RecordingJobRequestId, CommandError> {
        let (
            timeline_id,
            timeline_name,
            timeline_revision,
            collection_id,
            collection_path,
            entries,
        ) = if self.action == RecordingJobAction::StartView {
            // A free-camera view capture is a first-class 1-take job: one
            // synthetic open-ended entry, no timeline, no mount. The client
            // supplies only a display label (entries[0].name = the active
            // rig/preset or "freefly"); everything else is authored here so a
            // stale client cannot smuggle plan fields into execution.
            if self.job_id.trim().is_empty() {
                return Err(command_error(
                    &ctx,
                    "StartView requires a stable RecordingJob id",
                ));
            }
            let label = self
                .entries
                .first()
                .map(|entry| entry.name.trim().to_owned())
                .filter(|name| !name.is_empty())
                .unwrap_or_else(|| "freefly".to_owned());
            let entry = ShotEntryPlan {
                entry_id: "view".to_owned(),
                shot_id: String::new(),
                name: label,
                shot_index: None,
                kind: crate::ShotKind::Static,
                target_name: String::new(),
                translation_speed_cm_s: 0.0,
                rotation_speed_deg_s: 0.0,
                hold_duration_ms: 0,
                travel_duration_ms: 0,
                direction: crate::ShotDirection::default(),
                next_take_number: 1,
                open_ended: true,
            };
            (
                String::new(),
                String::new(),
                0,
                String::new(),
                Vec::new(),
                vec![entry],
            )
        } else if self.action == RecordingJobAction::Start {
            if self.capture_context.is_none() {
                return Err(command_error(
                    &ctx,
                    "Start requires editorial capture context",
                ));
            }
            if self.timeline_id.trim().is_empty() {
                return Err(CommandError {
                    tx: ctx.tx().to_string(),
                    command_id: ctx.command_id.to_string(),
                    message: "Start requires a persistent Timeline id".to_owned(),
                });
            }
            if self.job_id.trim().is_empty() {
                return Err(CommandError {
                    tx: ctx.tx().to_string(),
                    command_id: ctx.command_id.to_string(),
                    message: "Start requires a stable RecordingJob id".to_owned(),
                });
            }
            let timeline_id = TimelineId::from(self.timeline_id);
            let current = ctx
                .exec_query_first(GetTimelinesByIds {
                    ids: vec![timeline_id.clone()],
                })?
                .ok_or_else(|| CommandError {
                    tx: ctx.tx().to_string(),
                    command_id: ctx.command_id.to_string(),
                    message: format!("Timeline {timeline_id} does not exist"),
                })?;
            let mut timeline = (*current).clone();
            let (collection_id, collection_path) = if self.collection_id.trim().is_empty() {
                (String::new(), Vec::new())
            } else {
                let membership_id = CollectionMembershipId::from(CollectionMembership::stable_id(
                    &self.collection_id,
                    timeline_id.as_ref(),
                ));
                if ctx
                    .exec_query_first(GetCollectionMembershipsByIds {
                        ids: vec![membership_id],
                    })?
                    .is_none()
                {
                    return Err(command_error(
                        &ctx,
                        "Timeline is not assigned to the selected collection",
                    ));
                }
                let collections = ctx
                    .exec_query(GetCollectionsByQuery(CollectionQuery::default()))?
                    .into_iter()
                    .map(|collection| collection.as_ref().clone())
                    .collect::<Vec<_>>();
                let path = resolve_collection_path(&self.collection_id, &collections)
                    .map_err(|error| command_error(&ctx, error))?;
                (self.collection_id.clone(), path)
            };
            let revision = timeline.revision;
            let name = timeline.name.clone();
            let shot_rows = ctx
                .exec_query(GetShotsByQuery(ShotQuery::default()))?
                .into_iter()
                .map(|shot| shot.as_ref().clone())
                .collect::<Vec<_>>();
            let timeline_changed = timeline.backfill_entry_parameters(&shot_rows);
            let mut entries = crate::resolve_timeline_plans(&timeline, &shot_rows)
                .map_err(|error| command_error(&ctx, error))?;
            if entries.is_empty() {
                return Err(command_error(&ctx, "Timeline has no shot entries"));
            }
            // Optional single-shot / subset scope: keep only the requested
            // entry ids, preserving timeline order. Unknown ids are an error so
            // a stale UI can't silently record the wrong thing.
            if !self.entry_ids.is_empty() {
                let wanted: std::collections::HashSet<&str> =
                    self.entry_ids.iter().map(String::as_str).collect();
                entries.retain(|entry| wanted.contains(entry.entry_id.as_str()));
                if entries.len() != self.entry_ids.len() {
                    return Err(command_error(
                        &ctx,
                        "one or more requested shot entries are not in this timeline",
                    ));
                }
            }
            let prior_jobs = ctx
                .exec_query(GetRecordingJobsByQuery(RecordingJobQuery::default()))?
                .into_iter()
                .map(|job| job.as_ref().clone())
                .collect::<Vec<_>>();
            for entry in &mut entries {
                entry.next_take_number = crate::next_take_number_for_entry(
                    &prior_jobs,
                    &collection_id,
                    timeline_id.as_ref(),
                    entry,
                );
            }
            if timeline_changed {
                ctx.emit_set(&timeline)?;
            }
            (
                timeline_id.to_string(),
                name,
                revision,
                collection_id,
                collection_path,
                entries,
            )
        } else {
            (
                String::new(),
                String::new(),
                0,
                String::new(),
                Vec::new(),
                self.entries,
            )
        };
        let id: RecordingJobRequestId = self.streamer_id.clone().into();
        let capture_context = self.capture_context.map(StoredCaptureContext::from);
        ctx.emit_set(&RecordingJobRequest {
            id: id.clone(),
            streamer_id: self.streamer_id.clone(),
            job_id: self.job_id.clone(),
            command_id: self.command_id,
            action: self.action.clone(),
            capture_context: capture_context.clone(),
            timeline_id: timeline_id.clone(),
            timeline_name: timeline_name.clone(),
            timeline_revision,
            collection_id: collection_id.clone(),
            collection_path: collection_path.clone(),
            entries: entries.clone(),
            preset_duration_ms: self.preset_duration_ms,
            translation_speed_cm_s: self.translation_speed_cm_s,
            rotation_speed_deg_s: self.rotation_speed_deg_s,
            requested_at_ms: self.requested_at_ms,
        })?;
        if self.action == RecordingJobAction::Start {
            ctx.emit_set(&RecordingJob {
                id: RecordingJobId::from(self.job_id.clone()),
                job_id: self.job_id,
                streamer_id: self.streamer_id,
                capture_context,
                timeline_id,
                timeline_name,
                timeline_revision,
                collection_id,
                collection_path,
                phase: RecordingJobPhase::Idle,
                pause_requested: false,
                entries,
                takes: Vec::new(),
                error: String::new(),
                started_at_ms: 0,
                updated_at_ms: self.requested_at_ms,
                elapsed_ms: 0,
                estimated_total_ms: 0,
                estimated_remaining_ms: 0,
            })?;
        }
        Ok(id)
    }
}

/// Recorder-side authoritative RecordingJob progress report.
#[myko_command(RecordingJobStatusId)]
pub struct SetRecordingJobStatus {
    pub streamer_id: String,
    #[serde(alias = "runId")]
    pub job_id: String,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    #[ts(type = "unknown")]
    pub capture_context: Option<crate::CaptureContext>,
    #[serde(default, alias = "shotListId")]
    pub timeline_id: String,
    #[serde(default, alias = "shotListName")]
    pub timeline_name: String,
    #[serde(default, alias = "timelineVersion", alias = "shotListVersion")]
    pub timeline_revision: u32,
    #[serde(default)]
    pub collection_id: String,
    #[serde(default)]
    pub collection_path: Vec<crate::CollectionPathSegment>,
    pub phase: RecordingJobPhase,
    #[serde(default)]
    pub pause_requested: bool,
    #[serde(default, alias = "shots", alias = "items")]
    pub entries: Vec<ShotEntryPlan>,
    #[serde(default)]
    pub index: u32,
    #[serde(default)]
    pub completed: u32,
    #[serde(default)]
    pub error: String,
    #[serde(default)]
    pub updated_at_ms: u64,
    #[serde(default)]
    pub elapsed_ms: u64,
    #[serde(default)]
    pub estimated_total_ms: u64,
    #[serde(default)]
    pub estimated_remaining_ms: u64,
    #[serde(default)]
    pub takes: Vec<Take>,
    #[serde(default)]
    pub started_at_ms: u64,
}

/// True when `candidate` says nothing `stored` did not already say, apart from
/// the clock.
///
/// The event store is append-only, so a publisher that republishes unchanged
/// state on a timer grows it forever. Comparing with the stored
/// `updated_at_ms` substituted in isolates exactly that case: every real
/// field still participates, so any actual change writes normally.
fn recording_job_status_is_heartbeat_only(
    candidate: &RecordingJobStatus,
    stored: &RecordingJobStatus,
) -> bool {
    let mut probe = candidate.clone();
    probe.updated_at_ms = stored.updated_at_ms;
    probe == *stored
}

/// Same test for the target summaries the capture bridge mirrors on a poll.
fn target_summary_is_heartbeat_only(
    candidate: &crate::FrameCaptureTargetSummary,
    stored: &crate::FrameCaptureTargetSummary,
) -> bool {
    let mut probe = candidate.clone();
    probe.updated_at_ms = stored.updated_at_ms;
    probe == *stored
}

impl CommandHandler for SetRecordingJobStatus {
    fn execute(self, ctx: CommandContext) -> Result<RecordingJobStatusId, CommandError> {
        let id: RecordingJobStatusId = self.streamer_id.clone().into();
        let mut takes = self.takes;
        let job_id = self.job_id.clone();
        let existing_job = if !job_id.trim().is_empty() {
            ctx.exec_query_first(GetRecordingJobsByIds {
                ids: vec![RecordingJobId::from(job_id.clone())],
            })?
        } else {
            None
        };
        if let Some(existing) = &existing_job {
            for take in &mut takes {
                let accepted = existing.takes.iter().any(|saved| {
                    saved.take_id == take.take_id
                        && saved.capture.as_ref().is_some_and(|capture| {
                            capture.creative_status == CreativeStatus::Accepted
                        })
                });
                if accepted {
                    if let Some(capture) = &mut take.capture {
                        capture.creative_status = CreativeStatus::Accepted;
                    }
                }
            }
        }
        let mirror_present = existing_job.is_some();
        let capture_context = self
            .capture_context
            .map(StoredCaptureContext::from)
            .or_else(|| existing_job.and_then(|job| job.capture_context.clone()));
        let status = RecordingJobStatus {
            id: id.clone(),
            streamer_id: self.streamer_id,
            job_id: job_id.clone(),
            capture_context,
            timeline_id: self.timeline_id,
            timeline_name: self.timeline_name,
            timeline_revision: self.timeline_revision,
            collection_id: self.collection_id,
            collection_path: self.collection_path,
            phase: self.phase,
            pause_requested: self.pause_requested,
            entries: self.entries,
            index: self.index,
            completed: self.completed,
            error: self.error,
            updated_at_ms: self.updated_at_ms,
            elapsed_ms: self.elapsed_ms,
            estimated_total_ms: self.estimated_total_ms,
            estimated_remaining_ms: self.estimated_remaining_ms,
            takes,
            started_at_ms: self.started_at_ms,
        };
        // A recorder heartbeat carrying no new state must not become an event.
        // The store is append-only, so republishing an unchanged job every few
        // seconds per streamer is pure growth: it was 99.9% of all events
        // written and ~190 MB/day, with nothing but `updated_at_ms` differing
        // between consecutive rows. Suppressing it also makes the field mean
        // what every reader already assumes — when this job last *changed* —
        // rather than when the recorder last spoke.
        //
        // This lives here rather than in the recorder because the cell owns
        // the durable store: any publisher, now or later, gets the same floor.
        let unchanged = ctx
            .exec_query_first(GetRecordingJobStatussByIds {
                ids: vec![id.clone()],
            })?
            .is_some_and(|existing| recording_job_status_is_heartbeat_only(&status, &existing));
        // An unchanged status still writes when the paired RecordingJob mirror
        // is missing, so a half-written pair always converges.
        if unchanged && (job_id.trim().is_empty() || mirror_present) {
            return Ok(id);
        }
        ctx.emit_set(&status)?;
        if !job_id.trim().is_empty() {
            ctx.emit_set(&RecordingJob {
                id: RecordingJobId::from(job_id.clone()),
                job_id,
                streamer_id: status.streamer_id.clone(),
                capture_context: status.capture_context.clone(),
                timeline_id: status.timeline_id.clone(),
                timeline_name: status.timeline_name.clone(),
                timeline_revision: status.timeline_revision,
                collection_id: status.collection_id.clone(),
                collection_path: status.collection_path.clone(),
                phase: status.phase.clone(),
                pause_requested: status.pause_requested,
                entries: status.entries.clone(),
                takes: status.takes.clone(),
                error: status.error.clone(),
                started_at_ms: status.started_at_ms,
                updated_at_ms: status.updated_at_ms,
                elapsed_ms: status.elapsed_ms,
                estimated_total_ms: status.estimated_total_ms,
                estimated_remaining_ms: status.estimated_remaining_ms,
            })?;
        }
        Ok(id)
    }
}

/// Mark a delivered Take's Capture as operator-accepted without mutating its
/// integrity or QC evidence. Later recorder status republishes preserve review.
#[myko_command(RecordingJobId)]
pub struct AcceptTake {
    pub job_id: String,
    pub take_id: String,
}

impl CommandHandler for AcceptTake {
    fn execute(self, ctx: CommandContext) -> Result<RecordingJobId, CommandError> {
        let id = RecordingJobId::from(self.job_id);
        let current = ctx
            .exec_query_first(GetRecordingJobsByIds {
                ids: vec![id.clone()],
            })?
            .ok_or_else(|| command_error(&ctx, format!("Recording job {id} does not exist")))?;
        let mut job = current.as_ref().clone();
        let take = job
            .takes
            .iter_mut()
            .find(|take| take.take_id == self.take_id)
            .ok_or_else(|| command_error(&ctx, "Take does not exist"))?;
        let Some(capture) = &mut take.capture else {
            return Err(command_error(&ctx, "Take has no Capture to accept"));
        };
        if take.state != TakeState::Completed
            || capture.delivery_status != DeliveryStatus::Delivered
        {
            return Err(command_error(
                &ctx,
                "Only delivered Captures can be accepted",
            ));
        }
        capture.creative_status = CreativeStatus::Accepted;
        ctx.emit_set(&job)?;
        Ok(id)
    }
}

/// Upsert the durable global camera settings. The client sends this whenever the
/// operator changes a control, and reads them back (`GetCamPrefsByQuery`) on connect
/// to restore the panel + re-apply to the freshly-launched pawn. Full struct each
/// time (last-write-wins) — simplest and the payload is tiny.
#[myko_command(CamPrefId)]
pub struct SetCamPref {
    pub focal: f32,
    pub aperture: f32,
    pub focus_method: String,
    pub focus_dist: f32,
    pub base_speed: f32,
    pub look_scale: f32,
    pub invert: bool,
    pub glide: bool,
    pub glide_secs: f32,
    pub motion_blur: f32,
    pub rail_speed: f32,
}

impl CommandHandler for SetCamPref {
    fn execute(self, ctx: CommandContext) -> Result<CamPrefId, CommandError> {
        let id = CamPref::row_id();
        let pref = CamPref {
            id: id.clone(),
            focal: self.focal,
            aperture: self.aperture,
            focus_method: self.focus_method,
            focus_dist: self.focus_dist,
            base_speed: self.base_speed,
            look_scale: self.look_scale,
            invert: self.invert,
            glide: self.glide,
            glide_secs: self.glide_secs,
            motion_blur: self.motion_blur,
            rail_speed: self.rail_speed,
        };
        ctx.emit_set(&pref)?;
        Ok(id)
    }
}

/// Persist one stream's authored Home camera pose and lens. The command only
/// updates server state; moving the live camera remains an explicit client action.
#[myko_command(CameraHomeId)]
pub struct SetCameraHome {
    pub stream_id: String,
    pub location_x: f32,
    pub location_y: f32,
    pub location_z: f32,
    pub rotation_pitch: f32,
    pub rotation_yaw: f32,
    pub rotation_roll: f32,
    pub focal_length: f32,
}

impl CommandHandler for SetCameraHome {
    fn execute(self, ctx: CommandContext) -> Result<CameraHomeId, CommandError> {
        if self.stream_id.trim().is_empty() {
            return Err(command_error(&ctx, "Camera Home requires a stream id"));
        }
        let values = [
            self.location_x,
            self.location_y,
            self.location_z,
            self.rotation_pitch,
            self.rotation_yaw,
            self.rotation_roll,
            self.focal_length,
        ];
        if values.iter().any(|value| !value.is_finite()) {
            return Err(command_error(&ctx, "Camera Home values must be finite"));
        }
        if !(1.0..=1000.0).contains(&self.focal_length) {
            return Err(command_error(
                &ctx,
                "Camera Home focal length must be between 1 and 1000 mm",
            ));
        }

        let id = CameraHome::row_id(&self.stream_id);
        ctx.emit_set(&CameraHome {
            id: id.clone(),
            stream_id: self.stream_id,
            location_x: self.location_x,
            location_y: self.location_y,
            location_z: self.location_z,
            rotation_pitch: self.rotation_pitch,
            rotation_yaw: self.rotation_yaw,
            rotation_roll: self.rotation_roll,
            focal_length: self.focal_length,
        })?;
        Ok(id)
    }
}

/// Set a stream's friendly DISPLAY name (from the cluster def's `previs.stream_name`,
/// written on previs launch, keyed by the StreamerId). Upsert / last-write-wins. The
/// StreamerId is unchanged, so duplicate names never collide.
#[myko_command(StreamId)]
pub struct SetStreamName {
    pub stream_id: StreamId,
    pub name: String,
}

impl CommandHandler for SetStreamName {
    fn execute(self, ctx: CommandContext) -> Result<StreamId, CommandError> {
        let id = self.stream_id.clone();
        let stream = Stream {
            id: id.clone(),
            name: self.name,
        };
        ctx.emit_set(&stream)?;
        Ok(id)
    }
}

/// Client-side record trigger: set the intent to record `streamer_id` (active on/off) +
/// camera metadata. The recorder service watches `RecordingRequest`, chooses the active
/// rig (rail/crane) or preset/shot as its human artifact label, and starts/stops the
/// server-side no-transcode capture. Keyed by streamer_id (one active recording per stream).
#[myko_command(RecordingRequestId)]
pub struct SetRecording {
    pub streamer_id: String,
    pub active: bool,
    #[serde(default)]
    pub capture_kind: crate::RecordingKind,
    #[serde(default)]
    pub rig: String,
    #[serde(default)]
    pub preset: String,
    #[serde(default)]
    pub stream_name: String,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub travel_direction: Option<ShotDirection>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub shot_index: Option<u32>,
    #[serde(default)]
    pub take_number: u32,
    #[serde(default, alias = "clipId", alias = "cueId")]
    pub entry_id: String,
    #[serde(default, alias = "shotListId")]
    pub timeline_id: String,
    #[serde(default, alias = "shotListName")]
    pub timeline_name: String,
    #[serde(default, alias = "shotListVersion")]
    pub timeline_revision: u32,
    #[serde(default)]
    pub collection_path: Vec<crate::CollectionPathSegment>,
    #[serde(default)]
    pub requested_at_ms: u64,
}
impl CommandHandler for SetRecording {
    fn execute(self, ctx: CommandContext) -> Result<RecordingRequestId, CommandError> {
        let id: RecordingRequestId = self.streamer_id.clone().into();
        let req = RecordingRequest {
            id: id.clone(),
            streamer_id: self.streamer_id,
            active: self.active,
            capture_kind: self.capture_kind,
            rig: self.rig,
            preset: self.preset,
            stream_name: self.stream_name,
            travel_direction: self.travel_direction,
            shot_index: self.shot_index,
            take_number: self.take_number,
            entry_id: self.entry_id,
            timeline_id: self.timeline_id,
            timeline_name: self.timeline_name,
            timeline_revision: self.timeline_revision,
            collection_path: self.collection_path,
            requested_at_ms: self.requested_at_ms,
        };
        ctx.emit_set(&req)?;
        Ok(id)
    }
}

/// Recorder-side status report for capture, finalization, and NAS delivery readiness.
#[myko_command(RecordingStatusId)]
pub struct SetRecordingStatus {
    pub streamer_id: String,
    pub state: RecordingState,
    #[serde(default)]
    pub file_name: String,
    /// Absolute NAS path, so the UI can name where the take is instead of
    /// inferring a location from a file name.
    #[serde(default)]
    pub nas_path: String,
    /// Dropbox path from the team root, once the destination is named.
    #[serde(default)]
    pub dropbox_path: String,
    #[serde(default)]
    pub error: String,
    #[serde(default)]
    pub started_at_ms: u64,
}
impl CommandHandler for SetRecordingStatus {
    fn execute(self, ctx: CommandContext) -> Result<RecordingStatusId, CommandError> {
        let id: RecordingStatusId = self.streamer_id.clone().into();
        let st = RecordingStatus {
            id: id.clone(),
            streamer_id: self.streamer_id,
            state: self.state,
            file_name: self.file_name,
            nas_path: self.nas_path,
            dropbox_path: self.dropbox_path,
            error: self.error,
            started_at_ms: self.started_at_ms,
        };
        ctx.emit_set(&st)?;
        Ok(id)
    }
}

/// Register (or refresh) my presence on a stream.
#[myko_command(ViewerId)]
pub struct JoinStream {
    pub stream_id: StreamId,
    pub viewer_id: String,
    pub name: String,
    pub color: String,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub identity_issuer: Option<String>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub identity_subject: Option<String>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub avatar_url: Option<String>,
}

impl CommandHandler for JoinStream {
    fn execute(self, ctx: CommandContext) -> Result<ViewerId, CommandError> {
        let client_id = ctx
            .client_id()
            .map(|id| myko::entities::client::ClientId::from(id.to_owned()));
        // One presence row per CONNECTION: a browser's tabs share the persisted
        // viewer_id but must each be their own presence + drive unit. Fall back to
        // viewer_id only if there's somehow no client id (not a live WS connection).
        let conn = client_id
            .as_ref()
            .map(|c| c.to_string())
            .unwrap_or_else(|| self.viewer_id.clone());
        let id = Viewer::row_id(&self.stream_id, &conn);
        let stream_id = self.stream_id.clone();
        let viewer = Viewer {
            id: id.clone(),
            stream_id: self.stream_id,
            viewer_id: self.viewer_id,
            name: self.name,
            color: self.color,
            identity_issuer: self.identity_issuer,
            identity_subject: self.identity_subject,
            avatar_url: self.avatar_url,
            cursor: None,
            // Explicit: a command's emit_set is not auto-stamped with the client id.
            client_id,
        };
        ctx.emit_set(&viewer)?;
        // Reconcile the lock against live presence: clear a departed holder's
        // stale lock, and if this leaves exactly one viewer, they drive.
        reconcile_control(&ctx, &stream_id)?;
        Ok(id)
    }
}

/// Remove this connection's presence from a stream during in-page navigation.
///
/// A full websocket disconnect is already cascade-cleaned by myko, but moving
/// between streams keeps that socket alive. Addressing the same connection-keyed
/// row as [`JoinStream`] prevents ghost viewers and immediately reconciles the old
/// room's wheel.
#[myko_command]
pub struct LeaveStream {
    pub stream_id: StreamId,
    pub viewer_id: String,
}

impl CommandHandler for LeaveStream {
    fn execute(self, ctx: CommandContext) -> Result<(), CommandError> {
        let conn = ctx
            .client_id()
            .map(|client_id| client_id.to_string())
            .unwrap_or(self.viewer_id);
        let id = Viewer::row_id(&self.stream_id, &conn);
        if let Some(viewer) = ctx.exec_report(GetViewerById { id })? {
            ctx.emit_del(&*viewer)?;
        }
        reconcile_control(&ctx, &self.stream_id)
    }
}

/// Move my cursor (frequent, cheap).
#[myko_command]
pub struct UpdateCursor {
    pub stream_id: StreamId,
    pub viewer_id: String,
    pub cursor: Option<(f32, f32)>,
}

impl CommandHandler for UpdateCursor {
    fn execute(self, ctx: CommandContext) -> Result<(), CommandError> {
        // Address my own per-connection presence row (see JoinStream).
        let conn = ctx
            .client_id()
            .map(|c| c.to_string())
            .unwrap_or_else(|| self.viewer_id.clone());
        let id = Viewer::row_id(&self.stream_id, &conn);
        if let Some(viewer) = ctx.exec_report(GetViewerById { id })? {
            let updated = Viewer {
                cursor: self.cursor,
                ..(*viewer).clone()
            };
            ctx.emit_set(&updated)?;
        }
        Ok(())
    }
}

/// Take the wheel (SET overwrites any current holder — takeover).
#[myko_command(ControlLockId)]
pub struct AcquireControl {
    pub stream_id: StreamId,
    pub viewer_id: String,
}

impl CommandHandler for AcquireControl {
    fn execute(self, ctx: CommandContext) -> Result<ControlLockId, CommandError> {
        let id = ControlLock::row_id(&self.stream_id);
        let lock = ControlLock {
            id: id.clone(),
            stream_id: self.stream_id,
            viewer_id: self.viewer_id,
            client_id: ctx
                .client_id()
                .map(|id| myko::entities::client::ClientId::from(id.to_owned())),
        };
        ctx.emit_set(&lock)?;
        Ok(id)
    }
}

/// Release the wheel — held by a PERSON (viewer_id), so any of their tabs may
/// release it (the request carries the caller's viewer_id).
#[myko_command]
pub struct ReleaseControl {
    pub stream_id: StreamId,
    pub viewer_id: String,
}

impl CommandHandler for ReleaseControl {
    fn execute(self, ctx: CommandContext) -> Result<(), CommandError> {
        let id = ControlLock::row_id(&self.stream_id);
        if let Some(lock) = ctx.exec_report(GetControlLockById { id })? {
            if lock.viewer_id == self.viewer_id {
                ctx.emit_del(&*lock)?;
            }
        }
        Ok(())
    }
}

/// Server-internal: reconcile a stream's control lock against live presence.
/// Idempotent and cheap, safe to run on every join/leave.
///
/// The wheel belongs to a PERSON (viewer_id) — a person's tabs share it (browser
/// focus ensures only the active tab actually sends input), so a lock is kept alive
/// as long as ANY connection of the holder is present.
///
/// 1. **Clear a departed holder's lock.** `ControlLock` is not reliably
///    cascade-deleted on disconnect, so a holder who has fully left (no connection
///    of theirs remains) can linger; delete it. App-level backstop for the cascade.
/// 2. **Sole person auto-holds.** If exactly one person is present (any number of
///    their tabs) and doesn't already hold a valid lock, hand them the wheel.
/// The people behind a stream's live connections, deduplicated, sorted.
///
/// Multiple tabs of one person are one person: the wheel belongs to a human,
/// not to a socket.
fn people_present(viewers: &[Arc<Viewer>]) -> Vec<&str> {
    let mut people: Vec<&str> = viewers.iter().map(|v| v.viewer_id.as_str()).collect();
    people.sort_unstable();
    people.dedup();
    people
}

/// Whether this presence row still has a browser behind it.
///
/// Liveness is a property of the connection, never of the stored row. A row
/// carries no proof of life: a tab that dies without a disconnect leaves it
/// behind, replay restores it at boot, and it then looks exactly like presence.
/// myko's `ClientStatus` answers from the live connection registry instead —
/// the same source its `ConnectedClients` view uses.
///
/// A row with no client id cannot be tied to a connection at all, so it cannot
/// be shown to be alive.
fn viewer_is_live(ctx: &CommandContext, viewer: &Viewer) -> Result<bool, CommandError> {
    let Some(client_id) = viewer.client_id.clone() else {
        return Ok(false);
    };
    Ok(ctx.exec_report(ClientStatus { client_id })?.online)
}

fn reconcile_control(ctx: &CommandContext, stream_id: &StreamId) -> Result<(), CommandError> {
    let stored: Vec<Arc<Viewer>> = ctx.exec_query(GetViewersByQuery(ViewerQuery {
        stream_id: Some(IdFilter::Eq(stream_id.clone())),
        ..Default::default()
    }))?;
    // Every decision below is about who is *here*, so ask the connections, not
    // the store. Without this a ghost keeps a departed holder's lock alive and
    // can even be handed the wheel as the "sole" person — the recurring
    // "Anonymous ... has the wheel".
    let mut viewers = Vec::with_capacity(stored.len());
    for viewer in stored {
        if viewer_is_live(ctx, &viewer)? {
            viewers.push(viewer);
        }
    }
    let id = ControlLock::row_id(stream_id);

    // (1) Drop a lock whose holding PERSON has left (no connection of theirs remains).
    if let Some(lock) = ctx.exec_report(GetControlLockById { id: id.clone() })? {
        let holder_present = viewers.iter().any(|v| v.viewer_id == lock.viewer_id);
        if !holder_present {
            ctx.emit_del(&*lock)?;
        }
    }

    // (2) Sole PERSON auto-holds (multiple tabs of one person count as one).
    let people = people_present(&viewers);
    if let [sole_vid] = people.as_slice() {
        let held_by_sole = ctx
            .exec_report(GetControlLockById { id: id.clone() })?
            .as_deref()
            .is_some_and(|l| l.viewer_id.as_str() == *sole_vid);
        if !held_by_sole {
            // Carry the lock's client_id on any one of that person's connections.
            let conn = viewers
                .iter()
                .find(|v| v.viewer_id.as_str() == *sole_vid)
                .expect("present");
            ctx.emit_set(&ControlLock {
                id,
                stream_id: stream_id.clone(),
                viewer_id: (*sole_vid).to_string(),
                client_id: conn.client_id.clone(),
            })?;
        }
    }
    Ok(())
}

/// Server-internal command emitted by the leave-reassign saga (server crate):
/// reconcile a stream's control lock against live presence. Exposed so the saga can emit it.
#[myko_command]
pub struct AutoAssignControl {
    pub stream_id: StreamId,
}

impl CommandHandler for AutoAssignControl {
    fn execute(self, ctx: CommandContext) -> Result<(), CommandError> {
        reconcile_control(&ctx, &self.stream_id)
    }
}

/// Submit a synchronized nDisplay frame capture.
///
/// The client supplies ONLY typed editorial context and a typed target
/// selection; it never talks to Pulse Cluster and never constructs artifact
/// paths. The off-browser bridge picks this request up, submits it with the
/// cluster credential, and mirrors authoritative status + typed receipts back
/// as `FrameCaptureStatus`.
#[myko_command(FrameCaptureRequestId)]
pub struct SubmitFrameCapture {
    pub streamer_id: String,
    pub capture_id: String,
    pub command_id: String,
    #[ts(type = "unknown")]
    pub capture_context: crate::CaptureContext,
    pub target: crate::FrameCaptureTarget,
    /// Explicit operator override; see FrameCaptureRequest::force.
    #[serde(default)]
    pub force: bool,
    #[serde(default)]
    pub requested_at_ms: u64,
}

impl CommandHandler for SubmitFrameCapture {
    fn execute(self, ctx: CommandContext) -> Result<FrameCaptureRequestId, CommandError> {
        if self.streamer_id.trim().is_empty() {
            return Err(command_error(&ctx, "Frame capture requires a stream"));
        }
        if self.capture_id.trim().is_empty() {
            return Err(command_error(&ctx, "Frame capture requires a capture id"));
        }
        if self.target.cluster_name.trim().is_empty() {
            return Err(command_error(
                &ctx,
                "Frame capture requires an explicit target cluster",
            ));
        }
        // Editorial identity must be explicit — never inferred from whatever
        // collection happens to be browsed. The typed context carries it.
        let id: FrameCaptureRequestId = self.streamer_id.clone().into();
        ctx.emit_set(&crate::FrameCaptureRequest {
            id: id.clone(),
            streamer_id: self.streamer_id,
            capture_id: self.capture_id,
            command_id: self.command_id,
            capture_context: self.capture_context.into(),
            target: self.target,
            force: self.force,
            requested_at_ms: self.requested_at_ms,
        })?;
        Ok(id)
    }
}

/// Bridge-owned authoritative status for one stream's frame capture.
#[myko_command(FrameCaptureStatusId)]
pub struct SetFrameCaptureStatus {
    pub streamer_id: String,
    #[serde(default)]
    pub capture_id: String,
    pub phase: crate::FrameCapturePhase,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    #[ts(type = "unknown")]
    pub capture_context: Option<crate::CaptureContext>,
    #[serde(default)]
    pub target: crate::FrameCaptureTarget,
    #[serde(default)]
    pub observed_generation: String,
    #[serde(default)]
    pub receipts: Vec<crate::FrameCaptureReceipt>,
    #[serde(default)]
    pub error: String,
    #[serde(default)]
    pub forceable: bool,
    #[serde(default)]
    pub updated_at_ms: u64,
}

impl CommandHandler for SetFrameCaptureStatus {
    fn execute(self, ctx: CommandContext) -> Result<FrameCaptureStatusId, CommandError> {
        let id: FrameCaptureStatusId = self.streamer_id.clone().into();
        ctx.emit_set(&crate::FrameCaptureStatus {
            id: id.clone(),
            streamer_id: self.streamer_id,
            capture_id: self.capture_id,
            phase: self.phase,
            capture_context: self.capture_context.map(StoredCaptureContext::from),
            target: self.target,
            observed_generation: self.observed_generation,
            receipts: self.receipts,
            error: self.error,
            forceable: self.forceable,
            updated_at_ms: self.updated_at_ms,
        })?;
        Ok(id)
    }
}

/// Bridge-mirrored summary of a cluster the operator may capture on. The
/// browser never queries Pulse Cluster, so these summaries are how the UI
/// learns valid targets at all.
#[myko_command(FrameCaptureTargetSummaryId)]
pub struct SetFrameCaptureTargetSummary {
    pub cluster_name: String,
    #[serde(default)]
    pub generation: String,
    #[serde(default)]
    pub capturable: bool,
    #[serde(default)]
    pub status: String,
    #[serde(default)]
    pub updated_at_ms: u64,
}

impl CommandHandler for SetFrameCaptureTargetSummary {
    fn execute(self, ctx: CommandContext) -> Result<FrameCaptureTargetSummaryId, CommandError> {
        if self.cluster_name.trim().is_empty() {
            return Err(command_error(
                &ctx,
                "Target summary requires a cluster name",
            ));
        }
        let id: FrameCaptureTargetSummaryId = self.cluster_name.clone().into();
        let summary = crate::FrameCaptureTargetSummary {
            id: id.clone(),
            cluster_name: self.cluster_name,
            generation: self.generation,
            capturable: self.capturable,
            status: self.status,
            updated_at_ms: self.updated_at_ms,
        };
        // The bridge mirrors every target on a poll loop, so most of these
        // carry the same generation, status and capturability as the row
        // already stored — only the clock moved. An append-only store must not
        // record that: it was ~60k events/day for ~18 real changes.
        let unchanged = ctx
            .exec_query_first(GetFrameCaptureTargetSummarysByIds {
                ids: vec![id.clone()],
            })?
            .is_some_and(|existing| target_summary_is_heartbeat_only(&summary, &existing));
        if unchanged {
            return Ok(id);
        }
        ctx.emit_set(&summary)?;
        Ok(id)
    }
}

#[cfg(test)]
mod discovery_tests {
    use super::*;
    use crate::shot::DEFAULT_SHOT_LIBRARY_ID;

    fn tuned_legacy_shot() -> Shot {
        Shot {
            id: ShotId::from("render-11:moving:Floor Dolly"),
            library_id: String::new(),
            streamer_id: "render-11".to_owned(),
            name: "Floor Dolly".to_owned(),
            kind: ShotKind::Moving,
            target_name: "Floor Dolly".to_owned(),
            translation_speed_cm_s: 17.0,
            rotation_speed_deg_s: 3.5,
            hold_duration_ms: 8_000,
            travel_duration_ms: 41_000,
            default_entry_mode: ShotEntryMode::Both,
            shot_index: 9,
        }
    }

    fn legacy_timeline(id: &str, streamer_id: &str, name: &str, shot_id: &str) -> Timeline {
        Timeline {
            id: TimelineId::from(id),
            library_id: String::new(),
            streamer_id: streamer_id.to_owned(),
            name: name.to_owned(),
            revision: 0,
            entries: if shot_id.is_empty() {
                Vec::new()
            } else {
                vec![ShotEntry {
                    shot_id: shot_id.to_owned(),
                    direction: ShotEntryMode::Forward,
                    ..Default::default()
                }]
            },
            sort_order: 0,
        }
    }

    #[test]
    fn discovery_creates_only_missing_targets_and_preserves_tuned_legacy_rows() {
        let existing = vec![tuned_legacy_shot()];
        let discovered = vec![
            ShotDiscovery {
                name: "Floor Dolly".to_owned(),
                kind: ShotKind::Moving,
                target_name: "Floor Dolly".to_owned(),
            },
            ShotDiscovery {
                name: "Hero Push Forward".to_owned(),
                kind: ShotKind::Moving,
                target_name: "Hero Push Forward".to_owned(),
            },
            ShotDiscovery {
                name: "Hero Push Forward".to_owned(),
                kind: ShotKind::Moving,
                target_name: "Hero Push Forward".to_owned(),
            },
        ];

        let created = discovered_shots_to_create(DEFAULT_SHOT_LIBRARY_ID, discovered, &existing);
        assert_eq!(created.len(), 1);
        assert_eq!(created[0].name, "Hero Push Forward");
        assert_eq!(created[0].shot_index, 10);
        assert_eq!(
            created[0].translation_speed_cm_s,
            DEFAULT_SHOT_TRANSLATION_SPEED_CM_S
        );
        assert_eq!(
            created[0].rotation_speed_deg_s,
            DEFAULT_SHOT_ROTATION_SPEED_DEG_S
        );
        assert_eq!(created[0].default_entry_mode, ShotEntryMode::Forward);
        assert_eq!(existing[0].translation_speed_cm_s, 17.0);
        assert_eq!(existing[0].rotation_speed_deg_s, 3.5);
    }

    #[test]
    fn discovery_reconciles_live_legacy_timeline_duplicates_without_losing_clips() {
        let render_shot = tuned_legacy_shot();
        let mut studio_shot = tuned_legacy_shot();
        studio_shot.id = ShotId::from("Studio A:moving:Floor Dolly");
        studio_shot.streamer_id = "Studio A".to_owned();
        let timelines = vec![
            legacy_timeline(
                "render-11:timeline:default",
                "render-11",
                "Default shot list",
                render_shot.id.as_ref(),
            ),
            legacy_timeline(
                "Studio A:timeline:default",
                "Studio A",
                "Default timeline",
                studio_shot.id.as_ref(),
            ),
            legacy_timeline(
                "timeline-old",
                "render-11",
                "Supercut",
                render_shot.id.as_ref(),
            ),
            Timeline {
                id: TimelineId::from("timeline-shared"),
                library_id: DEFAULT_SHOT_LIBRARY_ID.to_owned(),
                streamer_id: "Studio A".to_owned(),
                name: " superCUT ".to_owned(),
                revision: 2,
                entries: vec![ShotEntry {
                    shot_id: studio_shot.id.to_string(),
                    direction: ShotEntryMode::Reverse,
                    ..Default::default()
                }],
                sort_order: 1,
            },
        ];

        let result = reconcile_timelines(
            DEFAULT_SHOT_LIBRARY_ID,
            &[render_shot, studio_shot],
            timelines,
        );

        assert_eq!(result.upserts.len(), 2);
        assert_eq!(result.deletes.len(), 3);
        let default = result
            .upserts
            .iter()
            .find(|timeline| timeline.id.to_string() == Timeline::shared_legacy_default_id())
            .expect("canonical shared default");
        assert_eq!(default.library_id, DEFAULT_SHOT_LIBRARY_ID);
        assert!(default.streamer_id.is_empty());
        assert_eq!(default.name, "Migrated timeline");
        assert_eq!(default.entries.len(), 1);
        assert_eq!(default.entries[0].shot_id, "Studio A:moving:Floor Dolly");
        assert_eq!(result.id_migrations.len(), 3);
        assert_eq!(
            result.id_migrations.get("render-11:timeline:default"),
            Some(&Timeline::shared_legacy_default_id())
        );

        let supercut = result
            .upserts
            .iter()
            .find(|timeline| timeline.id.as_ref() == "timeline-shared")
            .expect("explicit shared timeline wins");
        assert_eq!(supercut.name, "superCUT");
        assert_eq!(supercut.revision, 2);
        assert_eq!(supercut.entries.len(), 2);
        assert_eq!(supercut.entries[0].direction, ShotEntryMode::Reverse);
        assert_eq!(supercut.entries[1].direction, ShotEntryMode::Forward);
        assert!(supercut
            .entries
            .iter()
            .all(|entry| !entry.entry_id.is_empty()));
    }

    #[test]
    fn reconciliation_is_idempotent_for_canonical_timelines() {
        let shot = tuned_legacy_shot();
        let mut timeline = Timeline {
            id: TimelineId::from("shared:timeline:supercut"),
            library_id: DEFAULT_SHOT_LIBRARY_ID.to_owned(),
            streamer_id: String::new(),
            name: "Supercut".to_owned(),
            revision: 1,
            entries: vec![ShotEntry {
                shot_id: shot.id.to_string(),
                direction: ShotEntryMode::Forward,
                ..Default::default()
            }],
            sort_order: 0,
        };
        timeline.normalize_entries();
        assert!(timeline.backfill_entry_parameters(std::slice::from_ref(&shot)));
        assert!(!timeline.backfill_entry_parameters(std::slice::from_ref(&shot)));
        let result = reconcile_timelines(DEFAULT_SHOT_LIBRARY_ID, &[shot], vec![timeline]);
        assert!(result.upserts.is_empty());
        assert!(result.deletes.is_empty());
    }

    #[test]
    fn reconciliation_snapshots_only_missing_legacy_clip_parameters() {
        let shot = tuned_legacy_shot();
        let mut timeline =
            legacy_timeline("shared:timeline:supercut", "", "Supercut", shot.id.as_ref());
        timeline.library_id = DEFAULT_SHOT_LIBRARY_ID.to_owned();
        timeline.entries[0].translation_speed_cm_s = Some(4.5);

        let result = reconcile_timelines(
            DEFAULT_SHOT_LIBRARY_ID,
            std::slice::from_ref(&shot),
            vec![timeline],
        );
        assert_eq!(result.upserts.len(), 1);
        let entry = &result.upserts[0].entries[0];
        assert_eq!(entry.translation_speed_cm_s, Some(4.5));
        assert_eq!(entry.rotation_speed_deg_s, Some(shot.rotation_speed_deg_s));
        assert_eq!(entry.hold_duration_ms, Some(shot.hold_duration_ms));
        assert_eq!(
            entry.travel_duration_ms,
            Some(shot.effective_travel_duration_ms())
        );
        assert_eq!(entry.shot_index, Some(shot.shot_index));
    }
}

#[cfg(test)]
mod heartbeat_suppression_tests {
    use super::*;
    use crate::recording_job_status::RecordingJobStatus;

    fn status() -> RecordingJobStatus {
        RecordingJobStatus {
            id: "render-13".into(),
            streamer_id: "render-13".to_owned(),
            job_id: "job-1".to_owned(),
            capture_context: None,
            timeline_id: "supercut".to_owned(),
            timeline_name: "Supercut".to_owned(),
            timeline_revision: 1,
            collection_id: String::new(),
            collection_path: Vec::new(),
            phase: RecordingJobPhase::Traveling,
            pause_requested: false,
            entries: Vec::new(),
            index: 0,
            completed: 0,
            error: String::new(),
            updated_at_ms: 1_000,
            elapsed_ms: 5_000,
            estimated_total_ms: 10_000,
            estimated_remaining_ms: 5_000,
            takes: Vec::new(),
            started_at_ms: 500,
        }
    }

    fn summary() -> crate::FrameCaptureTargetSummary {
        crate::FrameCaptureTargetSummary {
            id: "0of12_rx11".into(),
            cluster_name: "0of12_rx11".to_owned(),
            generation: "3350".to_owned(),
            capturable: false,
            status: "stopped".to_owned(),
            updated_at_ms: 1_000,
        }
    }

    #[test]
    fn a_newer_clock_alone_is_not_a_change() {
        let stored = status();
        let mut republished = stored.clone();
        republished.updated_at_ms = 9_999;
        assert!(recording_job_status_is_heartbeat_only(
            &republished,
            &stored
        ));
    }

    #[test]
    fn real_progress_still_writes() {
        let stored = status();
        for mutate in [
            (|s: &mut RecordingJobStatus| s.phase = RecordingJobPhase::Complete)
                as fn(&mut RecordingJobStatus),
            |s: &mut RecordingJobStatus| s.elapsed_ms = 6_000,
            |s: &mut RecordingJobStatus| s.completed = 1,
            |s: &mut RecordingJobStatus| s.error = "disk full".to_owned(),
            |s: &mut RecordingJobStatus| s.pause_requested = true,
            |s: &mut RecordingJobStatus| s.estimated_remaining_ms = 4_000,
        ] {
            let mut candidate = stored.clone();
            candidate.updated_at_ms = 9_999;
            mutate(&mut candidate);
            assert!(
                !recording_job_status_is_heartbeat_only(&candidate, &stored),
                "a changed field must still be written"
            );
        }
    }

    #[test]
    fn target_summaries_follow_the_same_rule() {
        let stored = summary();
        let mut polled = stored.clone();
        polled.updated_at_ms = 9_999;
        assert!(target_summary_is_heartbeat_only(&polled, &stored));

        for mutate in [
            (|s: &mut crate::FrameCaptureTargetSummary| s.capturable = true)
                as fn(&mut crate::FrameCaptureTargetSummary),
            |s: &mut crate::FrameCaptureTargetSummary| s.generation = "3351".to_owned(),
            |s: &mut crate::FrameCaptureTargetSummary| s.status = "running".to_owned(),
        ] {
            let mut candidate = stored.clone();
            candidate.updated_at_ms = 9_999;
            mutate(&mut candidate);
            assert!(!target_summary_is_heartbeat_only(&candidate, &stored));
        }
    }
}

#[cfg(test)]
mod presence_tests {
    use std::sync::Arc;

    use super::people_present;
    use crate::viewer::Viewer;

    fn viewer(viewer_id: &str, client: &str) -> Arc<Viewer> {
        Arc::new(Viewer {
            id: format!("s1:{client}").into(),
            stream_id: "s1".into(),
            viewer_id: viewer_id.to_owned(),
            name: "Anonymous Cheetah".to_owned(),
            color: "#F87171".to_owned(),
            identity_issuer: None,
            identity_subject: None,
            avatar_url: None,
            cursor: None,
            client_id: Some(client.to_owned().into()),
        })
    }

    #[test]
    fn tabs_of_one_person_are_one_person() {
        // Two connections, one human — the wheel belongs to the human.
        let viewers = vec![viewer("max", "conn-a"), viewer("max", "conn-b")];
        assert_eq!(people_present(&viewers), vec!["max"]);
    }

    #[test]
    fn distinct_people_are_counted_separately_and_sorted() {
        let viewers = vec![
            viewer("zoe", "conn-c"),
            viewer("max", "conn-a"),
            viewer("max", "conn-b"),
        ];
        assert_eq!(people_present(&viewers), vec!["max", "zoe"]);
    }

    #[test]
    fn a_stream_whose_connections_all_died_has_nobody_present() {
        // reconcile_control filters to live rows before asking this, so the
        // ghost case arrives here as an empty list: no sole holder, and the
        // departed holder's lock is dropped rather than kept alive by a row
        // nobody is behind.
        assert!(people_present(&[]).is_empty());
    }
}