feat(api): owner tags and filtered movie views (#70)

.sqlx/query-0e5facf744fb0766253f352f51dafe5fecb377c6fd6cbeb6c8feb41f8e11ef77.json

@@ -0,0 +1,32 @@
+{
+  "db_name": "SQLite",
+  "query": "SELECT id AS \"id!: i64\", name AS \"name!: String\", ntfy_topic AS \"ntfy_topic!: String\" FROM owners ORDER BY name",
+  "describe": {
+    "columns": [
+      {
+        "name": "id!: i64",
+        "ordinal": 0,
+        "type_info": "Integer"
+      },
+      {
+        "name": "name!: String",
+        "ordinal": 1,
+        "type_info": "Text"
+      },
+      {
+        "name": "ntfy_topic!: String",
+        "ordinal": 2,
+        "type_info": "Text"
+      }
+    ],
+    "parameters": {
+      "Right": 0
+    },
+    "nullable": [
+      true,
+      false,
+      false
+    ]
+  },
+  "hash": "0e5facf744fb0766253f352f51dafe5fecb377c6fd6cbeb6c8feb41f8e11ef77"
+}

.sqlx/query-1cecbfbb3c147a5d4b41bde85e34ea34d757f7220d4a734c51dd48f7b2347371.json

@@ -0,0 +1,12 @@
+{
+  "db_name": "SQLite",
+  "query": "INSERT INTO owners (name, ntfy_topic) VALUES (?, ?)",
+  "describe": {
+    "columns": [],
+    "parameters": {
+      "Right": 2
+    },
+    "nullable": []
+  },
+  "hash": "1cecbfbb3c147a5d4b41bde85e34ea34d757f7220d4a734c51dd48f7b2347371"
+}

.sqlx/query-1d3d03463784627ec59f56a14129ba08b24efbcc69c7f115533c98b309805c71.json

@@ -0,0 +1,12 @@
+{
+  "db_name": "SQLite",
+  "query": "INSERT INTO title_owners (title_kind, title_id, owner_id) VALUES ('movie', ?, ?) ON CONFLICT DO NOTHING",
+  "describe": {
+    "columns": [],
+    "parameters": {
+      "Right": 2
+    },
+    "nullable": []
+  },
+  "hash": "1d3d03463784627ec59f56a14129ba08b24efbcc69c7f115533c98b309805c71"
+}

.sqlx/query-4da547256a5cf365948bdebaeca7353d0ef17c94710a47c5a2fcace1e77d14f7.json

@@ -0,0 +1,12 @@
+{
+  "db_name": "SQLite",
+  "query": "DELETE FROM title_owners WHERE title_kind = 'movie' AND title_id = ? AND owner_id = ?",
+  "describe": {
+    "columns": [],
+    "parameters": {
+      "Right": 2
+    },
+    "nullable": []
+  },
+  "hash": "4da547256a5cf365948bdebaeca7353d0ef17c94710a47c5a2fcace1e77d14f7"
+}

.sqlx/query-66f37c5521b20548add7978ebd371ea76a77ea85e3cb7c58db93016660b0394c.json

@@ -0,0 +1,86 @@
+{
+  "db_name": "SQLite",
+  "query": "SELECT m.id AS \"id!: i64\", m.tmdb_id AS \"tmdb_id!: i64\", m.title AS \"title!: String\", m.year, m.original_language, m.root_id AS \"root_id!: i64\", m.wanted AS \"wanted!: bool\", m.overrides AS \"overrides!: serde_json::Value\", m.state AS \"state!: String\", m.blocked AS \"blocked!: bool\", m.search_attempts AS \"search_attempts!: i64\", m.last_searched_at FROM movies m JOIN title_owners t ON t.title_kind = 'movie' AND t.title_id = m.id WHERE t.owner_id = ? ORDER BY m.title, m.year, m.id",
+  "describe": {
+    "columns": [
+      {
+        "name": "id!: i64",
+        "ordinal": 0,
+        "type_info": "Integer"
+      },
+      {
+        "name": "tmdb_id!: i64",
+        "ordinal": 1,
+        "type_info": "Integer"
+      },
+      {
+        "name": "title!: String",
+        "ordinal": 2,
+        "type_info": "Text"
+      },
+      {
+        "name": "year",
+        "ordinal": 3,
+        "type_info": "Integer"
+      },
+      {
+        "name": "original_language",
+        "ordinal": 4,
+        "type_info": "Text"
+      },
+      {
+        "name": "root_id!: i64",
+        "ordinal": 5,
+        "type_info": "Integer"
+      },
+      {
+        "name": "wanted!: bool",
+        "ordinal": 6,
+        "type_info": "Integer"
+      },
+      {
+        "name": "overrides!: serde_json::Value",
+        "ordinal": 7,
+        "type_info": "Text"
+      },
+      {
+        "name": "state!: String",
+        "ordinal": 8,
+        "type_info": "Text"
+      },
+      {
+        "name": "blocked!: bool",
+        "ordinal": 9,
+        "type_info": "Integer"
+      },
+      {
+        "name": "search_attempts!: i64",
+        "ordinal": 10,
+        "type_info": "Integer"
+      },
+      {
+        "name": "last_searched_at",
+        "ordinal": 11,
+        "type_info": "Text"
+      }
+    ],
+    "parameters": {
+      "Right": 1
+    },
+    "nullable": [
+      false,
+      false,
+      false,
+      true,
+      true,
+      false,
+      false,
+      false,
+      false,
+      false,
+      false,
+      true
+    ]
+  },
+  "hash": "66f37c5521b20548add7978ebd371ea76a77ea85e3cb7c58db93016660b0394c"
+}

.sqlx/query-82d8b805d307bcc1a202a5edc73e599a0ef3c8d075c7cce8832d581954167dc0.json

@@ -0,0 +1,12 @@
+{
+  "db_name": "SQLite",
+  "query": "UPDATE owners SET name = ?, ntfy_topic = ? WHERE id = ?",
+  "describe": {
+    "columns": [],
+    "parameters": {
+      "Right": 3
+    },
+    "nullable": []
+  },
+  "hash": "82d8b805d307bcc1a202a5edc73e599a0ef3c8d075c7cce8832d581954167dc0"
+}

.sqlx/query-8ce7b19c7e6524f4d14c92401d4f20c69e83f05f9acb568bc42270c7753b74cd.json

@@ -0,0 +1,32 @@
+{
+  "db_name": "SQLite",
+  "query": "SELECT o.id AS \"id!: i64\", o.name AS \"name!: String\", o.ntfy_topic AS \"ntfy_topic!: String\"\n           FROM owners o\n           JOIN title_owners t ON t.owner_id = o.id\n           WHERE t.title_kind = 'movie' AND t.title_id = ?\n           ORDER BY o.name",
+  "describe": {
+    "columns": [
+      {
+        "name": "id!: i64",
+        "ordinal": 0,
+        "type_info": "Integer"
+      },
+      {
+        "name": "name!: String",
+        "ordinal": 1,
+        "type_info": "Text"
+      },
+      {
+        "name": "ntfy_topic!: String",
+        "ordinal": 2,
+        "type_info": "Text"
+      }
+    ],
+    "parameters": {
+      "Right": 1
+    },
+    "nullable": [
+      true,
+      false,
+      false
+    ]
+  },
+  "hash": "8ce7b19c7e6524f4d14c92401d4f20c69e83f05f9acb568bc42270c7753b74cd"
+}

.sqlx/query-bb612df5646dba7ce195aa0a00068df7fa42de0b4322c78cc4cf43a695d16e54.json

@@ -0,0 +1,20 @@
+{
+  "db_name": "SQLite",
+  "query": "SELECT EXISTS(SELECT 1 FROM owners WHERE id = ?) AS 'exists!: bool'",
+  "describe": {
+    "columns": [
+      {
+        "name": "exists!: bool",
+        "ordinal": 0,
+        "type_info": "Integer"
+      }
+    ],
+    "parameters": {
+      "Right": 1
+    },
+    "nullable": [
+      false
+    ]
+  },
+  "hash": "bb612df5646dba7ce195aa0a00068df7fa42de0b4322c78cc4cf43a695d16e54"
+}

.sqlx/query-bcec423ef801043a1d1934bb241ffeeabfaa027360e3317100687fbf0334a95d.json

@@ -0,0 +1,32 @@
+{
+  "db_name": "SQLite",
+  "query": "SELECT id AS \"id!: i64\", name AS \"name!: String\", ntfy_topic AS \"ntfy_topic!: String\" FROM owners WHERE id = ?",
+  "describe": {
+    "columns": [
+      {
+        "name": "id!: i64",
+        "ordinal": 0,
+        "type_info": "Integer"
+      },
+      {
+        "name": "name!: String",
+        "ordinal": 1,
+        "type_info": "Text"
+      },
+      {
+        "name": "ntfy_topic!: String",
+        "ordinal": 2,
+        "type_info": "Text"
+      }
+    ],
+    "parameters": {
+      "Right": 1
+    },
+    "nullable": [
+      false,
+      false,
+      false
+    ]
+  },
+  "hash": "bcec423ef801043a1d1934bb241ffeeabfaa027360e3317100687fbf0334a95d"
+}

.sqlx/query-d5c820b9b78c9901212359a2d57264382b7eb64bceef2c27d7c1de7c19974477.json

@@ -0,0 +1,12 @@
+{
+  "db_name": "SQLite",
+  "query": "DELETE FROM owners WHERE id = ?",
+  "describe": {
+    "columns": [],
+    "parameters": {
+      "Right": 1
+    },
+    "nullable": []
+  },
+  "hash": "d5c820b9b78c9901212359a2d57264382b7eb64bceef2c27d7c1de7c19974477"
+}

crates/arr-api/src/lib.rs

@@ -8,6 +8,7 @@
 
 mod health;
 mod movies;
+mod owners;
 mod search;
 mod state;
 
@@ -20,6 +21,7 @@ use utoipa_scalar::{Scalar, Servable};
 
 pub use health::{Check, Health, HealthReport, Status};
 pub use movies::{Accepted, AttentionQueues, CreateMovie, ErrorBody, Movie, Release, UpdateMovie};
+pub use owners::{CreateOwner, Owner, UpdateOwner};
 pub use search::{ClassifiedRelease, SearchResponse};
 pub use state::{AppState, MovieCommand, Upstreams, DEFAULT_TMDB_URL};
 
@@ -42,6 +44,7 @@ pub const DOCS_PATH: &str = "/api/docs";
     tags(
         (name = "system", description = "Service health and metadata"),
         (name = "movies", description = "Movie library and actions"),
+        (name = "owners", description = "Owner tags and filtered views (DESIGN.md §4.3)"),
         (name = "search", description = "Unified title and release search")
     ),
 )]
@@ -57,6 +60,10 @@ fn api_router() -> OpenApiRouter<AppState> {
         .routes(routes!(movies::releases))
         .routes(routes!(movies::grab))
         .routes(routes!(movies::attention))
+        .routes(routes!(movies::list_owners))
+        .routes(routes!(movies::tag_owner, movies::untag_owner))
+        .routes(routes!(owners::list, owners::create))
+        .routes(routes!(owners::get, owners::update, owners::delete))
         .routes(routes!(search::search))
         .routes(routes!(search::releases))
 }

crates/arr-api/src/movies.rs

@@ -1,10 +1,11 @@
-use axum::extract::{Path, State};
+use axum::extract::{Path, Query, State};
 use axum::http::StatusCode;
 use axum::response::{IntoResponse, Response};
 use axum::Json;
 use serde::{Deserialize, Serialize};
-use utoipa::ToSchema;
+use utoipa::{IntoParams, ToSchema};
 
+use crate::owners::Owner;
 use crate::state::{AppState, MovieCommand};
 
 #[derive(Debug, Clone, Serialize, ToSchema)]
@@ -84,6 +85,7 @@ pub struct ErrorBody {
 #[derive(Debug)]
 pub enum ApiError {
     NotFound,
+    OwnerNotFound,
     Conflict(String),
     Invalid(String),
     Unavailable,
@@ -94,6 +96,7 @@ impl IntoResponse for ApiError {
     fn into_response(self) -> Response {
         let (status, error) = match self {
             Self::NotFound => (StatusCode::NOT_FOUND, "movie not found".to_string()),
+            Self::OwnerNotFound => (StatusCode::NOT_FOUND, "owner not found".to_string()),
             Self::Conflict(error) => (StatusCode::CONFLICT, error),
             Self::Invalid(error) => (StatusCode::UNPROCESSABLE_ENTITY, error),
             Self::Unavailable => (
@@ -150,7 +153,7 @@ fn validate_overrides(value: &serde_json::Value) -> Result<(), ApiError> {
     Ok(())
 }
 
-fn pool(state: &AppState) -> Result<&sqlx::SqlitePool, ApiError> {
+pub(crate) fn pool(state: &AppState) -> Result<&sqlx::SqlitePool, ApiError> {
     state
         .database()
         .map(arr_db::Db::pool)
@@ -163,18 +166,34 @@ async fn load_movie(state: &AppState, id: i64) -> Result<Movie, ApiError> {
         .await?)
 }
 
+#[derive(Debug, Deserialize, IntoParams)]
+pub struct ListMoviesQuery {
+    /// Restrict to movies tagged with this owner (DESIGN.md §4.3).
+    pub owner_id: Option<i64>,
+}
+
 #[utoipa::path(
     get, path = "/api/movies", tag = "movies",
+    params(ListMoviesQuery),
     responses(
         (status = 200, body = [Movie]),
         (status = 500, body = ErrorBody),
         (status = 503, body = ErrorBody)
     )
 )]
-pub async fn list(State(state): State<AppState>) -> Result<Json<Vec<Movie>>, ApiError> {
+pub async fn list(
-    let movies = sqlx::query_as!(Movie, r#"SELECT id AS "id!: i64", tmdb_id AS "tmdb_id!: i64", title AS "title!: String", year, original_language, root_id AS "root_id!: i64", wanted AS "wanted!: bool", overrides AS "overrides!: serde_json::Value", state AS "state!: String", blocked AS "blocked!: bool", search_attempts AS "search_attempts!: i64", last_searched_at FROM movies ORDER BY title, year, id"#)
+    State(state): State<AppState>,
-        .fetch_all(pool(&state)?)
+    Query(query): Query<ListMoviesQuery>,
-        .await?;
+) -> Result<Json<Vec<Movie>>, ApiError> {
+    let movies = if let Some(owner_id) = query.owner_id {
+        sqlx::query_as!(Movie, r#"SELECT m.id AS "id!: i64", m.tmdb_id AS "tmdb_id!: i64", m.title AS "title!: String", m.year, m.original_language, m.root_id AS "root_id!: i64", m.wanted AS "wanted!: bool", m.overrides AS "overrides!: serde_json::Value", m.state AS "state!: String", m.blocked AS "blocked!: bool", m.search_attempts AS "search_attempts!: i64", m.last_searched_at FROM movies m JOIN title_owners t ON t.title_kind = 'movie' AND t.title_id = m.id WHERE t.owner_id = ? ORDER BY m.title, m.year, m.id"#, owner_id)
+            .fetch_all(pool(&state)?)
+            .await?
+    } else {
+        sqlx::query_as!(Movie, r#"SELECT id AS "id!: i64", tmdb_id AS "tmdb_id!: i64", title AS "title!: String", year, original_language, root_id AS "root_id!: i64", wanted AS "wanted!: bool", overrides AS "overrides!: serde_json::Value", state AS "state!: String", blocked AS "blocked!: bool", search_attempts AS "search_attempts!: i64", last_searched_at FROM movies ORDER BY title, year, id"#)
+            .fetch_all(pool(&state)?)
+            .await?
+    };
     Ok(Json(movies))
 }
 
@@ -407,6 +426,93 @@ pub async fn attention(State(state): State<AppState>) -> Result<Json<AttentionQu
     }))
 }
 
+#[utoipa::path(
+    get, path = "/api/movies/{movie_id}/owners", tag = "movies",
+    params(("movie_id" = i64, Path, description = "Movie row id")),
+    responses(
+        (status = 200, body = [Owner]),
+        (status = 404, body = ErrorBody),
+        (status = 500, body = ErrorBody),
+        (status = 503, body = ErrorBody)
+    )
+)]
+pub async fn list_owners(
+    State(state): State<AppState>,
+    Path(id): Path<i64>,
+) -> Result<Json<Vec<Owner>>, ApiError> {
+    load_movie(&state, id).await?;
+    let owners = sqlx::query_as!(
+        Owner,
+        r#"SELECT o.id AS "id!: i64", o.name AS "name!: String", o.ntfy_topic AS "ntfy_topic!: String"
+           FROM owners o
+           JOIN title_owners t ON t.owner_id = o.id
+           WHERE t.title_kind = 'movie' AND t.title_id = ?
+           ORDER BY o.name"#,
+        id
+    )
+    .fetch_all(pool(&state)?)
+    .await?;
+    Ok(Json(owners))
+}
+
+#[utoipa::path(
+    put, path = "/api/movies/{movie_id}/owners/{owner_id}", tag = "movies",
+    params(("movie_id" = i64, Path), ("owner_id" = i64, Path)),
+    responses(
+        (status = 204),
+        (status = 404, body = ErrorBody),
+        (status = 500, body = ErrorBody),
+        (status = 503, body = ErrorBody)
+    )
+)]
+pub async fn tag_owner(
+    State(state): State<AppState>,
+    Path((movie_id, owner_id)): Path<(i64, i64)>,
+) -> Result<StatusCode, ApiError> {
+    load_movie(&state, movie_id).await?;
+    let owner_exists = sqlx::query_scalar!(
+        "SELECT EXISTS(SELECT 1 FROM owners WHERE id = ?) AS 'exists!: bool'",
+        owner_id
+    )
+    .fetch_one(pool(&state)?)
+    .await?;
+    if !owner_exists {
+        return Err(ApiError::OwnerNotFound);
+    }
+    sqlx::query!(
+        "INSERT INTO title_owners (title_kind, title_id, owner_id) VALUES ('movie', ?, ?) ON CONFLICT DO NOTHING",
+        movie_id, owner_id
+    )
+    .execute(pool(&state)?)
+    .await?;
+    Ok(StatusCode::NO_CONTENT)
+}
+
+#[utoipa::path(
+    delete, path = "/api/movies/{movie_id}/owners/{owner_id}", tag = "movies",
+    params(("movie_id" = i64, Path), ("owner_id" = i64, Path)),
+    responses(
+        (status = 204),
+        (status = 404, body = ErrorBody),
+        (status = 500, body = ErrorBody),
+        (status = 503, body = ErrorBody)
+    )
+)]
+pub async fn untag_owner(
+    State(state): State<AppState>,
+    Path((movie_id, owner_id)): Path<(i64, i64)>,
+) -> Result<StatusCode, ApiError> {
+    load_movie(&state, movie_id).await?;
+    sqlx::query!(
+        "DELETE FROM title_owners WHERE title_kind = 'movie' AND title_id = ? AND owner_id = ?",
+        movie_id,
+        owner_id
+    )
+    .execute(pool(&state)?)
+    .await?;
+    Ok(StatusCode::NO_CONTENT)
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;

crates/arr-api/src/owners.rs

@@ -0,0 +1,362 @@
+//! Owner tags: a many-to-many label on titles that drives filtered views and
+//! notification routing (`DESIGN.md` §4.3). Movies only for now — series
+//! arrives with TV (§13 phase 6); the join table already carries `title_kind`
+//! so nothing here changes shape when that lands.
+
+use axum::extract::{Path, State};
+use axum::http::StatusCode;
+use axum::Json;
+use serde::{Deserialize, Serialize};
+use utoipa::ToSchema;
+
+use crate::movies::{pool, ApiError, ErrorBody};
+use crate::state::AppState;
+
+#[derive(Debug, Clone, Serialize, ToSchema)]
+pub struct Owner {
+    pub id: i64,
+    pub name: String,
+    pub ntfy_topic: String,
+}
+
+#[derive(Debug, Deserialize, ToSchema)]
+pub struct CreateOwner {
+    pub name: String,
+    pub ntfy_topic: String,
+}
+
+#[derive(Debug, Deserialize, ToSchema)]
+pub struct UpdateOwner {
+    pub name: Option<String>,
+    pub ntfy_topic: Option<String>,
+}
+
+async fn load_owner(state: &AppState, id: i64) -> Result<Owner, ApiError> {
+    sqlx::query_as!(
+        Owner,
+        r#"SELECT id AS "id!: i64", name AS "name!: String", ntfy_topic AS "ntfy_topic!: String" FROM owners WHERE id = ?"#,
+        id
+    )
+    .fetch_optional(pool(state)?)
+    .await?
+    .ok_or(ApiError::OwnerNotFound)
+}
+
+#[utoipa::path(
+    get, path = "/api/owners", tag = "owners",
+    responses(
+        (status = 200, body = [Owner]),
+        (status = 500, body = ErrorBody),
+        (status = 503, body = ErrorBody)
+    )
+)]
+pub async fn list(State(state): State<AppState>) -> Result<Json<Vec<Owner>>, ApiError> {
+    let owners = sqlx::query_as!(
+        Owner,
+        r#"SELECT id AS "id!: i64", name AS "name!: String", ntfy_topic AS "ntfy_topic!: String" FROM owners ORDER BY name"#
+    )
+    .fetch_all(pool(&state)?)
+    .await?;
+    Ok(Json(owners))
+}
+
+#[utoipa::path(
+    post, path = "/api/owners", tag = "owners", request_body = CreateOwner,
+    responses(
+        (status = 201, body = Owner),
+        (status = 409, body = ErrorBody),
+        (status = 422, body = ErrorBody),
+        (status = 500, body = ErrorBody),
+        (status = 503, body = ErrorBody)
+    )
+)]
+pub async fn create(
+    State(state): State<AppState>,
+    Json(input): Json<CreateOwner>,
+) -> Result<(StatusCode, Json<Owner>), ApiError> {
+    let name = input.name.trim();
+    let ntfy_topic = input.ntfy_topic.trim();
+    if name.is_empty() || ntfy_topic.is_empty() {
+        return Err(ApiError::Invalid("name and ntfy_topic are required".into()));
+    }
+    let result = sqlx::query!(
+        "INSERT INTO owners (name, ntfy_topic) VALUES (?, ?)",
+        name,
+        ntfy_topic
+    )
+    .execute(pool(&state)?)
+    .await?;
+    Ok((
+        StatusCode::CREATED,
+        Json(load_owner(&state, result.last_insert_rowid()).await?),
+    ))
+}
+
+#[utoipa::path(
+    get, path = "/api/owners/{owner_id}", tag = "owners",
+    params(("owner_id" = i64, Path, description = "Owner row id")),
+    responses(
+        (status = 200, body = Owner),
+        (status = 404, body = ErrorBody),
+        (status = 500, body = ErrorBody),
+        (status = 503, body = ErrorBody)
+    )
+)]
+pub async fn get(
+    State(state): State<AppState>,
+    Path(id): Path<i64>,
+) -> Result<Json<Owner>, ApiError> {
+    Ok(Json(load_owner(&state, id).await?))
+}
+
+#[utoipa::path(
+    patch, path = "/api/owners/{owner_id}", tag = "owners", request_body = UpdateOwner,
+    params(("owner_id" = i64, Path, description = "Owner row id")),
+    responses(
+        (status = 200, body = Owner),
+        (status = 404, body = ErrorBody),
+        (status = 409, body = ErrorBody),
+        (status = 422, body = ErrorBody),
+        (status = 500, body = ErrorBody),
+        (status = 503, body = ErrorBody)
+    )
+)]
+pub async fn update(
+    State(state): State<AppState>,
+    Path(id): Path<i64>,
+    Json(input): Json<UpdateOwner>,
+) -> Result<Json<Owner>, ApiError> {
+    let current = load_owner(&state, id).await?;
+    let name = input.name.unwrap_or(current.name);
+    let ntfy_topic = input.ntfy_topic.unwrap_or(current.ntfy_topic);
+    let name = name.trim();
+    let ntfy_topic = ntfy_topic.trim();
+    if name.is_empty() || ntfy_topic.is_empty() {
+        return Err(ApiError::Invalid("name and ntfy_topic are required".into()));
+    }
+    sqlx::query!(
+        "UPDATE owners SET name = ?, ntfy_topic = ? WHERE id = ?",
+        name,
+        ntfy_topic,
+        id
+    )
+    .execute(pool(&state)?)
+    .await?;
+    Ok(Json(load_owner(&state, id).await?))
+}
+
+#[utoipa::path(
+    delete, path = "/api/owners/{owner_id}", tag = "owners",
+    params(("owner_id" = i64, Path, description = "Owner row id")),
+    responses(
+        (status = 204),
+        (status = 404, body = ErrorBody),
+        (status = 500, body = ErrorBody),
+        (status = 503, body = ErrorBody)
+    )
+)]
+pub async fn delete(
+    State(state): State<AppState>,
+    Path(id): Path<i64>,
+) -> Result<StatusCode, ApiError> {
+    let result = sqlx::query!("DELETE FROM owners WHERE id = ?", id)
+        .execute(pool(&state)?)
+        .await?;
+    if result.rows_affected() == 0 {
+        return Err(ApiError::OwnerNotFound);
+    }
+    Ok(StatusCode::NO_CONTENT)
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::{router, Upstreams};
+
+    async fn application() -> (tempfile::TempDir, AppState, String) {
+        let dir = tempfile::tempdir().expect("tempdir");
+        let database = arr_db::Db::connect(dir.path().join("arr.db"))
+            .await
+            .expect("connect database");
+        database.migrate().await.expect("migrate database");
+        let state = AppState::new(Upstreams::new(
+            "http://127.0.0.1:1".into(),
+            "http://127.0.0.1:1".into(),
+        ))
+        .expect("state")
+        .with_database(database);
+        let listener = tokio::net::TcpListener::bind("127.0.0.1:0")
+            .await
+            .expect("bind");
+        let address = listener.local_addr().expect("address");
+        let app = router(state.clone());
+        tokio::spawn(async move { axum::serve(listener, app).await.expect("serve") });
+        (dir, state, format!("http://{address}"))
+    }
+
+    async fn create_owner(base: &str, name: &str) -> serde_json::Value {
+        let response = reqwest::Client::new()
+            .post(format!("{base}/api/owners"))
+            .json(&serde_json::json!({ "name": name, "ntfy_topic": format!("topic-{name}") }))
+            .send()
+            .await
+            .expect("create owner");
+        assert_eq!(response.status(), StatusCode::CREATED);
+        response.json().await.expect("owner json")
+    }
+
+    async fn create_movie(base: &str, tmdb_id: i64, root_id: i64) -> serde_json::Value {
+        let response = reqwest::Client::new()
+            .post(format!("{base}/api/movies"))
+            .json(&serde_json::json!({
+                "tmdb_id": tmdb_id, "title": "Dune Part Two", "year": 2024,
+                "original_language": "en", "root_id": root_id,
+            }))
+            .send()
+            .await
+            .expect("create movie");
+        assert_eq!(response.status(), StatusCode::CREATED);
+        response.json().await.expect("movie json")
+    }
+
+    #[tokio::test]
+    async fn crud_round_trips_an_owner() {
+        let (_dir, _state, base) = application().await;
+        let owner = create_owner(&base, "miguel").await;
+        let id = owner["id"].as_i64().expect("id");
+
+        let updated: serde_json::Value = reqwest::Client::new()
+            .patch(format!("{base}/api/owners/{id}"))
+            .json(&serde_json::json!({ "ntfy_topic": "new-topic" }))
+            .send()
+            .await
+            .expect("update")
+            .json()
+            .await
+            .expect("updated json");
+        assert_eq!(updated["name"], "miguel");
+        assert_eq!(updated["ntfy_topic"], "new-topic");
+
+        let deleted = reqwest::Client::new()
+            .delete(format!("{base}/api/owners/{id}"))
+            .send()
+            .await
+            .expect("delete");
+        assert_eq!(deleted.status(), StatusCode::NO_CONTENT);
+        assert_eq!(
+            reqwest::get(format!("{base}/api/owners/{id}"))
+                .await
+                .expect("get deleted")
+                .status(),
+            StatusCode::NOT_FOUND
+        );
+    }
+
+    #[tokio::test]
+    async fn a_movie_with_two_owners_appears_in_both_filtered_views_but_once_on_disk() {
+        let (_dir, state, base) = application().await;
+        let pool = state.database().expect("database").pool();
+        let root_id: i64 = sqlx::query_scalar("SELECT id FROM roots WHERE audience = 'main'")
+            .fetch_one(pool)
+            .await
+            .expect("main root");
+
+        let movie = create_movie(&base, 693_134, root_id).await;
+        let movie_id = movie["id"].as_i64().expect("id");
+        let alice = create_owner(&base, "alice").await;
+        let bob = create_owner(&base, "bob").await;
+        let alice_id = alice["id"].as_i64().expect("id");
+        let bob_id = bob["id"].as_i64().expect("id");
+
+        for owner_id in [alice_id, bob_id] {
+            let response = reqwest::Client::new()
+                .put(format!("{base}/api/movies/{movie_id}/owners/{owner_id}"))
+                .send()
+                .await
+                .expect("tag");
+            assert_eq!(response.status(), StatusCode::NO_CONTENT);
+        }
+
+        for owner_id in [alice_id, bob_id] {
+            let filtered: Vec<serde_json::Value> =
+                reqwest::get(format!("{base}/api/movies?owner_id={owner_id}"))
+                    .await
+                    .expect("filtered list")
+                    .json()
+                    .await
+                    .expect("filtered json");
+            assert_eq!(filtered.len(), 1, "owner {owner_id}");
+            assert_eq!(filtered[0]["id"], movie_id);
+        }
+
+        let on_disk: i64 = sqlx::query_scalar("SELECT count(*) FROM movies WHERE id = ?")
+            .bind(movie_id)
+            .fetch_one(pool)
+            .await
+            .expect("count");
+        assert_eq!(on_disk, 1, "the title exists once regardless of tags");
+
+        let tags: Vec<serde_json::Value> =
+            reqwest::get(format!("{base}/api/movies/{movie_id}/owners"))
+                .await
+                .expect("owners of movie")
+                .json()
+                .await
+                .expect("owners json");
+        assert_eq!(tags.len(), 2);
+
+        let response = reqwest::Client::new()
+            .delete(format!("{base}/api/movies/{movie_id}/owners/{alice_id}"))
+            .send()
+            .await
+            .expect("untag");
+        assert_eq!(response.status(), StatusCode::NO_CONTENT);
+
+        let alice_view: Vec<serde_json::Value> =
+            reqwest::get(format!("{base}/api/movies?owner_id={alice_id}"))
+                .await
+                .expect("alice view")
+                .json()
+                .await
+                .expect("alice json");
+        assert!(alice_view.is_empty());
+
+        let bob_view: Vec<serde_json::Value> =
+            reqwest::get(format!("{base}/api/movies?owner_id={bob_id}"))
+                .await
+                .expect("bob view")
+                .json()
+                .await
+                .expect("bob json");
+        assert_eq!(bob_view.len(), 1);
+    }
+
+    #[tokio::test]
+    async fn tagging_a_missing_movie_or_owner_is_not_found() {
+        let (_dir, state, base) = application().await;
+        let owner = create_owner(&base, "alice").await;
+        let owner_id = owner["id"].as_i64().expect("id");
+
+        let response = reqwest::Client::new()
+            .put(format!("{base}/api/movies/999/owners/{owner_id}"))
+            .send()
+            .await
+            .expect("tag missing movie");
+        assert_eq!(response.status(), StatusCode::NOT_FOUND);
+
+        let pool = state.database().expect("database").pool();
+        let root_id: i64 = sqlx::query_scalar("SELECT id FROM roots WHERE audience = 'main'")
+            .fetch_one(pool)
+            .await
+            .expect("main root");
+        let movie = create_movie(&base, 693_134, root_id).await;
+        let movie_id = movie["id"].as_i64().expect("id");
+
+        let response = reqwest::Client::new()
+            .put(format!("{base}/api/movies/{movie_id}/owners/999"))
+            .send()
+            .await
+            .expect("tag missing owner");
+        assert_eq!(response.status(), StatusCode::NOT_FOUND);
+    }
+}