michal/tit

Browse tree · Show commit · Download archive

Diff

48fc552adfd5 → d6b66655459b

@@ -320,3 +320,17 @@
 sessions, CSRF checks, and forms that operate without JavaScript. Read the
 [issue workflow architectural decision record](docs/adr/0017-issue-workflow.md)
 for the permission and event contracts.
+
+## Milestone 4.3 gate
+
+Run the repository watch gate:
+
+```text
+./scripts/check-m4-3
+```
+
+This command tests granular push, issue, and pull-request preferences, the
+“everything” selection, stable watch IDs, permission checks, removal, private
+preference handling, CSRF checks, and forms that operate without JavaScript.
+Read the [repository watches architectural decision record](docs/adr/0018-repository-watches.md)
+for the storage and privacy contracts.

@@ -1,0 +1,70 @@
+# Architectural decision record 0018: Repository watches
+
+Status: Accepted
+
+Date: 2026-07-22
+
+## Context
+
+An account must be able to select repository activity for later personalized
+feeds. The first release does not have an inbox or a background mail process.
+Watch data is private preference data. It must not appear in the public
+repository event stream.
+
+## Decision
+
+Store one optional watch record for an account and repository. Give the record
+a random, non-reassignable ID. Store three independent Boolean preferences:
+
+- pushes;
+- issues;
+- pull requests.
+
+All three selected values mean “everything.” No selected value means “stop
+watching.” Delete the watch record in that case. Do not store a record that has
+no selected activity.
+
+An active account can change a watch only when it can read the repository. Use
+the common repository visibility and role data for this check. Anonymous users
+can read the watch page for a public repository, but the page does not show
+private preference data.
+
+Use one upsert for a preference change. Keep the watch ID and creation time when
+the account changes its selections. Add an index on account and repository for
+the personalized feed query in Milestone 4.4.
+
+Do not append a repository event for a watch change. A public event would expose
+private preference data and could cause a watch to notify itself. The watch row
+is the canonical preference state.
+
+The Web page uses three normal HTML select controls. The page and form operate
+without JavaScript. The form requires an active session and a matching CSRF
+value.
+
+## Failure and threat cases
+
+A database constraint rejects an invalid ID, a non-Boolean value, a repeated
+account and repository pair, or a row with no selected activity. A suspended
+account and an account that cannot read a private repository cannot read or
+change its watch through the service.
+
+A private watch is selected only after the service checks current repository
+access. A subsequent feed request must check access again. Thus, a retained
+watch cannot restore access after a role is removed.
+
+## Evidence
+
+Storage tests select everything, change to issue-only activity, preserve the
+watch ID and creation time, reject unauthorized accounts, reject an empty row,
+and delete the row when all selections are clear. They also prove that watch
+changes do not append repository events.
+
+A production HTTP test reads the anonymous page, sets all preferences with an
+authenticated form, reads the selected values, and clears the watch. The common
+Web workflow test continues to verify CSRF rejection.
+
+## Consequences
+
+Milestone 4.4 can select watched event kinds without a schema change. A later
+inbox or mail process can use the same preferences, but this milestone does not
+add either process.

@@ -1,0 +1,6 @@
+#!/bin/sh
+set -eu
+
+./scripts/check
+cargo test --locked --release --test sqlite runs_the_issue_workflow_with_atomic_events_and_repository_roles
+cargo test --locked --release --test public_routes runs_the_complete_issue_workflow_without_javascript

@@ -12,8 +12,8 @@
 use crate::store::{IssueDetail, StoreError};
 
 use super::{
-    CSRF_COOKIE, RequestActor, RequestId, SESSION_COOKIE, WebState, cookie, login_job,
-    login_redirect, parse_named_form, render, render_error,
+    CSRF_COOKIE, RequestActor, RequestId, WebState, authenticate_mutation, cookie,
+    parse_named_form, render, render_error,
 };
 
 const MAX_ISSUE_REQUEST_BYTES: usize = 300 * 1024;
@@ -327,34 +327,6 @@
         Ok(()) => issue_redirect(&redirect_owner, &redirect_repository, redirect_number),
         Err(error) => issue_mutation_error(error, &request_id.0),
     }
-}
-
-async fn authenticate_mutation(
-    state: WebState,
-    headers: &HeaderMap,
-    submitted_csrf: &str,
-    request_id: &str,
-) -> Result<String, Response> {
-    let Some(session_token) = cookie(headers, SESSION_COOKIE) else {
-        return Err(login_redirect(false));
-    };
-    let Some(csrf) = cookie(headers, CSRF_COOKIE) else {
-        return Err(login_redirect(true));
-    };
-    if submitted_csrf != csrf {
-        return Err(render_error(
-            StatusCode::FORBIDDEN,
-            request_id,
-            "Forbidden",
-            "The request is not authorized.",
-        ));
-    }
-    login_job(state, move |login| {
-        login.authenticate(&session_token, Some(&csrf))
-    })
-    .await
-    .map(|session| session.username)
-    .map_err(|_| login_redirect(true))
 }
 
 async fn issue_job<T: Send + 'static>(

@@ -1,5 +1,6 @@
 mod issues;
 mod public;
+mod watches;
 
 use std::net::SocketAddr;
 use std::path::PathBuf;
@@ -25,6 +26,7 @@
 use crate::repository::{RepositoryService, RepositoryServiceError};
 use crate::session::{SessionError, WebLoginService};
 use crate::store::StoreError;
+use crate::watch::WatchService;
 
 use self::public::PublicWeb;
 
@@ -45,6 +47,7 @@
     login: Option<WebLoginService>,
     repositories: Option<RepositoryService>,
     issues: Option<IssueService>,
+    watches: Option<WatchService>,
     secure_cookies: bool,
 }
 
@@ -78,6 +81,7 @@
                 login: None,
                 repositories: None,
                 issues: None,
+                watches: None,
                 secure_cookies: false,
             },
         )
@@ -114,6 +118,7 @@
         let public = PublicWeb::open(config, Arc::clone(&jobs))?;
         let repositories = RepositoryService::new(public.database(), public.repository_root());
         let issues = IssueService::new(public.database());
+        let watches = WatchService::new(public.database());
         Self::start_with_state(
             address,
             WebState {
@@ -124,6 +129,7 @@
                 login: Some(login),
                 repositories: Some(repositories),
                 issues: Some(issues),
+                watches: Some(watches),
                 secure_cookies,
             },
         )
@@ -168,18 +174,19 @@
         login: None,
         repositories: None,
         issues: None,
+        watches: None,
         secure_cookies: false,
     })
 }
 
 fn router_with_state(state: WebState) -> Router {
-    let repository_routes =
-        issues::routes()
-            .merge(public::routes())
-            .layer(middleware::from_fn_with_state(
-                state.clone(),
-                repository_actor,
-            ));
+    let repository_routes = watches::routes()
+        .merge(issues::routes())
+        .merge(public::routes())
+        .layer(middleware::from_fn_with_state(
+            state.clone(),
+            repository_actor,
+        ));
     Router::new()
         .route("/", get(home))
         .route("/go", get(go_to_repository))
@@ -800,6 +807,34 @@
     .map_err(|_| {
         RepositoryServiceError::Store(StoreError::Integrity("repository worker failed".to_owned()))
     })?
+}
+
+async fn authenticate_mutation(
+    state: WebState,
+    headers: &HeaderMap,
+    submitted_csrf: &str,
+    request_id: &str,
+) -> Result<String, Response> {
+    let Some(session_token) = cookie(headers, SESSION_COOKIE) else {
+        return Err(login_redirect(false));
+    };
+    let Some(csrf) = cookie(headers, CSRF_COOKIE) else {
+        return Err(login_redirect(true));
+    };
+    if submitted_csrf != csrf {
+        return Err(render_error(
+            StatusCode::FORBIDDEN,
+            request_id,
+            "Forbidden",
+            "The request is not authorized.",
+        ));
+    }
+    login_job(state, move |login| {
+        login.authenticate(&session_token, Some(&csrf))
+    })
+    .await
+    .map(|session| session.username)
+    .map_err(|_| login_redirect(true))
 }
 
 fn parse_named_form(

@@ -1,0 +1,209 @@
+use askama::Template;
+use axum::Router;
+use axum::body::Bytes;
+use axum::extract::{DefaultBodyLimit, Extension, Path, State};
+use axum::http::{HeaderMap, StatusCode, header};
+use axum::response::Response;
+use axum::routing::get;
+use serde::Deserialize;
+
+use crate::store::{StoreError, WatchPreferences};
+use crate::watch::WatchError;
+
+use super::{
+    CSRF_COOKIE, RequestActor, RequestId, WebState, authenticate_mutation, cookie,
+    parse_named_form, render, render_error,
+};
+
+pub(super) fn routes() -> Router<WebState> {
+    Router::new().route(
+        "/{owner}/{repository}/watch",
+        get(watch_page)
+            .post(set_watch)
+            .layer(DefaultBodyLimit::max(2048)),
+    )
+}
+
+async fn watch_page(
+    State(state): State<WebState>,
+    Extension(request_id): Extension<RequestId>,
+    Extension(actor): Extension<RequestActor>,
+    Path(path): Path<RepositoryPath>,
+    headers: HeaderMap,
+) -> Response {
+    let Some(service) = state.watches.clone() else {
+        return watch_internal(&request_id.0);
+    };
+    let owner = path.owner;
+    let repository = path.repository;
+    let authenticated = actor.0.is_some();
+    let result = watch_job(state, move || {
+        service.get(&owner, &repository, actor.0.as_deref())
+    })
+    .await;
+    match result {
+        Ok((record, watch)) => {
+            let csrf = cookie(&headers, CSRF_COOKIE).unwrap_or_default();
+            let preferences = watch
+                .map(|record| WatchPreferences {
+                    pushes: record.pushes,
+                    issues: record.issues,
+                    pull_requests: record.pull_requests,
+                })
+                .unwrap_or(WatchPreferences {
+                    pushes: false,
+                    issues: false,
+                    pull_requests: false,
+                });
+            render(
+                StatusCode::OK,
+                &WatchTemplate {
+                    request_id: &request_id.0,
+                    owner: &record.owner,
+                    repository: &record.slug,
+                    csrf: &csrf,
+                    can_change: authenticated && !csrf.is_empty(),
+                    pushes: preferences.pushes,
+                    issues: preferences.issues,
+                    pull_requests: preferences.pull_requests,
+                    watching: preferences.any(),
+                },
+            )
+        }
+        Err(error) => watch_error(error, &request_id.0),
+    }
+}
+
+async fn set_watch(
+    State(state): State<WebState>,
+    Extension(request_id): Extension<RequestId>,
+    Path(path): Path<RepositoryPath>,
+    headers: HeaderMap,
+    body: Bytes,
+) -> Response {
+    let fields = match parse_named_form(
+        &headers,
+        &body,
+        &["csrf", "pushes", "issues", "pull-requests"],
+    ) {
+        Ok(fields) => fields,
+        Err(()) => return watch_bad_request(&request_id.0),
+    };
+    let preferences = match (
+        preference(&fields[1]),
+        preference(&fields[2]),
+        preference(&fields[3]),
+    ) {
+        (Ok(pushes), Ok(issues), Ok(pull_requests)) => WatchPreferences {
+            pushes,
+            issues,
+            pull_requests,
+        },
+        _ => return watch_bad_request(&request_id.0),
+    };
+    let actor =
+        match authenticate_mutation(state.clone(), &headers, &fields[0], &request_id.0).await {
+            Ok(actor) => actor,
+            Err(response) => return response,
+        };
+    let Some(service) = state.watches.clone() else {
+        return watch_internal(&request_id.0);
+    };
+    let owner = path.owner.clone();
+    let repository = path.repository.clone();
+    let result = watch_job(state, move || {
+        service.set(&owner, &repository, &actor, preferences)
+    })
+    .await;
+    match result {
+        Ok(_) => Response::builder()
+            .status(StatusCode::SEE_OTHER)
+            .header(
+                header::LOCATION,
+                format!("/{}/{}/watch", path.owner, path.repository),
+            )
+            .header(header::CACHE_CONTROL, "no-store")
+            .body(axum::body::Body::empty())
+            .expect("the watch redirect is valid"),
+        Err(error) => watch_error(error, &request_id.0),
+    }
+}
+
+async fn watch_job<T: Send + 'static>(
+    state: WebState,
+    operation: impl FnOnce() -> Result<T, WatchError> + Send + 'static,
+) -> Result<T, WatchError> {
+    let permit = state.jobs.acquire_owned().await.map_err(|_| {
+        WatchError::Store(StoreError::Integrity(
+            "watch worker pool is unavailable".to_owned(),
+        ))
+    })?;
+    tokio::task::spawn_blocking(move || {
+        let _permit = permit;
+        operation()
+    })
+    .await
+    .map_err(|_| WatchError::Store(StoreError::Integrity("watch worker failed".to_owned())))?
+}
+
+fn preference(value: &str) -> Result<bool, ()> {
+    match value {
+        "0" => Ok(false),
+        "1" => Ok(true),
+        _ => Err(()),
+    }
+}
+
+fn watch_error(error: WatchError, request_id: &str) -> Response {
+    match error {
+        WatchError::Auth(_)
+        | WatchError::RepositoryName(_)
+        | WatchError::Store(StoreError::RepositoryNotFound(_, _) | StoreError::WatchDenied) => {
+            render_error(
+                StatusCode::NOT_FOUND,
+                request_id,
+                "Not found",
+                "The repository was not found.",
+            )
+        }
+        _ => watch_internal(request_id),
+    }
+}
+
+fn watch_bad_request(request_id: &str) -> Response {
+    render_error(
+        StatusCode::BAD_REQUEST,
+        request_id,
+        "Watch error",
+        "The watch request is not valid.",
+    )
+}
+
+fn watch_internal(request_id: &str) -> Response {
+    render_error(
+        StatusCode::INTERNAL_SERVER_ERROR,
+        request_id,
+        "Watch error",
+        "The watch request could not be completed.",
+    )
+}
+
+#[derive(Deserialize)]
+struct RepositoryPath {
+    owner: String,
+    repository: String,
+}
+
+#[derive(Template)]
+#[template(path = "watch.html")]
+struct WatchTemplate<'a> {
+    request_id: &'a str,
+    owner: &'a str,
+    repository: &'a str,
+    csrf: &'a str,
+    can_change: bool,
+    pushes: bool,
+    issues: bool,
+    pull_requests: bool,
+    watching: bool,
+}

@@ -25,6 +25,7 @@
 #[allow(dead_code, reason = "the server uses only part of the shared SSH API")]
 mod ssh;
 mod store;
+mod watch;
 
 use std::process::ExitCode;
 use std::{io, io::Write};

@@ -1,0 +1,22 @@
+CREATE TABLE watch (
+    id TEXT PRIMARY KEY
+        CHECK (
+            length(id) = 32
+            AND id = lower(id)
+            AND id NOT GLOB '*[^0-9a-f]*'
+        ),
+    repository_id TEXT NOT NULL
+        REFERENCES repository (id) ON DELETE RESTRICT,
+    account_id INTEGER NOT NULL
+        REFERENCES account (id) ON DELETE RESTRICT,
+    pushes INTEGER NOT NULL CHECK (pushes IN (0, 1)),
+    issues INTEGER NOT NULL CHECK (issues IN (0, 1)),
+    pull_requests INTEGER NOT NULL CHECK (pull_requests IN (0, 1)),
+    created_at INTEGER NOT NULL CHECK (created_at >= 0),
+    updated_at INTEGER NOT NULL CHECK (updated_at >= created_at),
+    UNIQUE (repository_id, account_id),
+    CHECK (pushes = 1 OR issues = 1 OR pull_requests = 1)
+) STRICT;
+
+CREATE INDEX watch_account_activity
+ON watch (account_id, repository_id);

@@ -11,7 +11,7 @@
 
 const BUSY_TIMEOUT: Duration = Duration::from_secs(5);
 const BUSY_TIMEOUT_MILLISECONDS: i64 = 5_000;
-const SCHEMA_VERSION: i64 = 12;
+const SCHEMA_VERSION: i64 = 13;
 #[allow(
     dead_code,
     reason = "the integration test imports this module without the CLI operation"
@@ -21,7 +21,7 @@
     dead_code,
     reason = "M1A proves migrations before the M2 server calls them"
 )]
-const MIGRATIONS: [&str; 12] = [
+const MIGRATIONS: [&str; 13] = [
     include_str!("migrations/001_initial.sql"),
     include_str!("migrations/002_state.sql"),
     include_str!("migrations/003_git_intents.sql"),
@@ -34,6 +34,7 @@
     include_str!("migrations/010_audit_history.sql"),
     include_str!("migrations/011_domain_events.sql"),
     include_str!("migrations/012_issues.sql"),
+    include_str!("migrations/013_watches.sql"),
 ];
 
 #[allow(
@@ -130,6 +131,8 @@
     IssueAssigneeState,
     #[error("issue assignee does not exist or cannot read the repository: {0}")]
     IssueAssigneeNotFound(String),
+    #[error("repository watch access is not authorized")]
+    WatchDenied,
 }
 
 pub(crate) struct Store {
@@ -1839,6 +1842,102 @@
         })
     }
 
+    pub(crate) fn watch(
+        &self,
+        owner: &str,
+        repository: &str,
+        actor: Option<&str>,
+    ) -> Result<(RepositoryRecord, Option<WatchRecord>), StoreError> {
+        let access = repository_issue_access(&self.connection, owner, repository, actor)?;
+        if !access.can_read() {
+            return Err(StoreError::WatchDenied);
+        }
+        let watch = match access.active_actor_id {
+            Some(actor_id) => self
+                .connection
+                .query_row(
+                    "SELECT id, pushes, issues, pull_requests, created_at, updated_at
+                 FROM watch WHERE repository_id = ?1 AND account_id = ?2",
+                    rusqlite::params![access.repository.id, actor_id],
+                    |row| {
+                        Ok(WatchRecord {
+                            id: row.get(0)?,
+                            pushes: row.get(1)?,
+                            issues: row.get(2)?,
+                            pull_requests: row.get(3)?,
+                            created_at: row.get(4)?,
+                            updated_at: row.get(5)?,
+                        })
+                    },
+                )
+                .optional()?,
+            None => None,
+        };
+        Ok((access.repository, watch))
+    }
+
+    pub(crate) fn set_watch(
+        &mut self,
+        owner: &str,
+        repository: &str,
+        actor: &str,
+        preferences: WatchPreferences,
+        changed_at: i64,
+    ) -> Result<Option<WatchRecord>, StoreError> {
+        let transaction = self
+            .connection
+            .transaction_with_behavior(TransactionBehavior::Immediate)?;
+        let access = repository_issue_access(&transaction, owner, repository, Some(actor))?;
+        let actor_id = access.active_actor_id.ok_or(StoreError::WatchDenied)?;
+        if !access.can_read() {
+            return Err(StoreError::WatchDenied);
+        }
+        if !preferences.any() {
+            transaction.execute(
+                "DELETE FROM watch WHERE repository_id = ?1 AND account_id = ?2",
+                rusqlite::params![access.repository.id, actor_id],
+            )?;
+            transaction.commit()?;
+            return Ok(None);
+        }
+        transaction.execute(
+            "INSERT INTO watch
+             (id, repository_id, account_id, pushes, issues, pull_requests,
+              created_at, updated_at)
+             VALUES (lower(hex(randomblob(16))), ?1, ?2, ?3, ?4, ?5, ?6, ?6)
+             ON CONFLICT (repository_id, account_id) DO UPDATE SET
+                 pushes = excluded.pushes,
+                 issues = excluded.issues,
+                 pull_requests = excluded.pull_requests,
+                 updated_at = excluded.updated_at",
+            rusqlite::params![
+                access.repository.id,
+                actor_id,
+                preferences.pushes,
+                preferences.issues,
+                preferences.pull_requests,
+                changed_at,
+            ],
+        )?;
+        let record = transaction.query_row(
+            "SELECT id, pushes, issues, pull_requests, created_at, updated_at
+             FROM watch WHERE repository_id = ?1 AND account_id = ?2",
+            rusqlite::params![access.repository.id, actor_id],
+            |row| {
+                Ok(WatchRecord {
+                    id: row.get(0)?,
+                    pushes: row.get(1)?,
+                    issues: row.get(2)?,
+                    pull_requests: row.get(3)?,
+                    created_at: row.get(4)?,
+                    updated_at: row.get(5)?,
+                })
+            },
+        )?;
+        transaction.commit()?;
+        Ok(Some(record))
+    }
+
     #[allow(
         dead_code,
         reason = "some integration tests compile storage without authorization"
@@ -2248,6 +2347,29 @@
     pub(crate) number: i64,
     pub(crate) actor: &'a str,
     pub(crate) changed_at: i64,
+}
+
+#[derive(Clone, Copy, Debug, Eq, PartialEq)]
+pub(crate) struct WatchPreferences {
+    pub(crate) pushes: bool,
+    pub(crate) issues: bool,
+    pub(crate) pull_requests: bool,
+}
+
+impl WatchPreferences {
+    pub(crate) fn any(self) -> bool {
+        self.pushes || self.issues || self.pull_requests
+    }
+}
+
+#[derive(Clone, Debug, Eq, PartialEq)]
+pub(crate) struct WatchRecord {
+    pub(crate) id: String,
+    pub(crate) pushes: bool,
+    pub(crate) issues: bool,
+    pub(crate) pull_requests: bool,
+    pub(crate) created_at: i64,
+    pub(crate) updated_at: i64,
 }
 
 pub(crate) struct GitOperationIntent<'a> {

@@ -1,0 +1,77 @@
+use std::path::{Path, PathBuf};
+use std::time::{SystemTime, UNIX_EPOCH};
+
+use thiserror::Error;
+
+use crate::auth::{AuthError, validate_username};
+use crate::domain::repository::{RepositoryNameError, validate_slug};
+use crate::store::{RepositoryRecord, Store, StoreError, WatchPreferences, WatchRecord};
+
+#[derive(Clone)]
+pub(crate) struct WatchService {
+    database: PathBuf,
+}
+
+impl WatchService {
+    pub(crate) fn new(database: &Path) -> Self {
+        Self {
+            database: database.to_owned(),
+        }
+    }
+
+    pub(crate) fn get(
+        &self,
+        owner: &str,
+        repository: &str,
+        actor: Option<&str>,
+    ) -> Result<(RepositoryRecord, Option<WatchRecord>), WatchError> {
+        validate_username(owner)?;
+        validate_slug(repository)?;
+        if let Some(actor) = actor {
+            validate_username(actor)?;
+        }
+        Store::open(&self.database)?
+            .watch(owner, repository, actor)
+            .map_err(Into::into)
+    }
+
+    pub(crate) fn set(
+        &self,
+        owner: &str,
+        repository: &str,
+        actor: &str,
+        preferences: WatchPreferences,
+    ) -> Result<Option<WatchRecord>, WatchError> {
+        validate(owner, repository, actor)?;
+        Store::open(&self.database)?
+            .set_watch(owner, repository, actor, preferences, timestamp()?)
+            .map_err(Into::into)
+    }
+}
+
+fn validate(owner: &str, repository: &str, actor: &str) -> Result<(), WatchError> {
+    validate_username(owner)?;
+    validate_slug(repository)?;
+    validate_username(actor)?;
+    Ok(())
+}
+
+fn timestamp() -> Result<i64, WatchError> {
+    let seconds = SystemTime::now()
+        .duration_since(UNIX_EPOCH)
+        .map_err(|_| WatchError::Clock)?
+        .as_secs();
+    i64::try_from(seconds).map_err(|_| WatchError::Clock)
+}
+
+#[derive(Debug, Error)]
+pub(crate) enum WatchError {
+    #[error(transparent)]
+    Auth(#[from] AuthError),
+    #[error(transparent)]
+    RepositoryName(#[from] RepositoryNameError),
+    #[error(transparent)]
+    Store(#[from] StoreError),
+    #[error("system time is not valid")]
+    Clock,
+}

@@ -7,6 +7,7 @@
       <a href="/{{ owner }}/{{ repository }}">Summary</a>
       <a href="/{{ owner }}/{{ repository }}/refs">Refs</a>
       <a href="/{{ owner }}/{{ repository }}/issues">Issues</a>
+      <a href="/{{ owner }}/{{ repository }}/watch">Watch</a>
       <a href="/{{ owner }}/{{ repository }}/atom.xml">Atom</a>
       <a href="/{{ owner }}/{{ repository }}/rss.xml">RSS</a>
     </nav>

@@ -7,6 +7,7 @@
       <a href="/{{ owner }}/{{ repository }}">Summary</a>
       <a href="/{{ owner }}/{{ repository }}/refs">Refs</a>
       <a href="/{{ owner }}/{{ repository }}/issues">Issues</a>
+      <a href="/{{ owner }}/{{ repository }}/watch">Watch</a>
       <a href="/{{ owner }}/{{ repository }}/atom.xml">Atom</a>
       <a href="/{{ owner }}/{{ repository }}/rss.xml">RSS</a>
     </nav>

@@ -7,6 +7,7 @@
       <a href="/{{ owner }}/{{ repository }}">Summary</a>
       <a href="/{{ owner }}/{{ repository }}/refs">Refs</a>
       <a href="/{{ owner }}/{{ repository }}/issues">Issues</a>
+      <a href="/{{ owner }}/{{ repository }}/watch">Watch</a>
       <a href="/{{ owner }}/{{ repository }}/atom.xml">Atom</a>
       <a href="/{{ owner }}/{{ repository }}/rss.xml">RSS</a>
       <a href="/{{ owner }}/{{ repository }}/search">Search</a>

@@ -1,0 +1,52 @@
+{% extends "base.html" %}
+{% block title %}Watch · {{ owner }}/{{ repository }} · tit{% endblock %}
+{% block content %}
+  <header class="repository-header">
+    <h1><a href="/{{ owner }}/{{ repository }}">{{ owner }}/{{ repository }}</a></h1>
+    <nav aria-label="Repository">
+      <a href="/{{ owner }}/{{ repository }}">Summary</a>
+      <a href="/{{ owner }}/{{ repository }}/refs">Refs</a>
+      <a href="/{{ owner }}/{{ repository }}/issues">Issues</a>
+      <a href="/{{ owner }}/{{ repository }}/watch">Watch</a>
+      <a href="/{{ owner }}/{{ repository }}/atom.xml">Atom</a>
+      <a href="/{{ owner }}/{{ repository }}/rss.xml">RSS</a>
+    </nav>
+  </header>
+
+  <h2>Watch preferences</h2>
+{% if watching %}
+  <p>You watch selected activity in this repository.</p>
+{% else %}
+  <p>You do not watch this repository.</p>
+{% endif %}
+{% if can_change %}
+  <form method="post" action="/{{ owner }}/{{ repository }}/watch">
+    <input type="hidden" name="csrf" value="{{ csrf }}">
+    <div class="field">
+      <label for="watch-pushes">Pushes</label>
+      <select id="watch-pushes" name="pushes">
+        <option value="0"{% if !pushes %} selected{% endif %}>Do not watch</option>
+        <option value="1"{% if pushes %} selected{% endif %}>Watch</option>
+      </select>
+    </div>
+    <div class="field">
+      <label for="watch-issues">Issues</label>
+      <select id="watch-issues" name="issues">
+        <option value="0"{% if !issues %} selected{% endif %}>Do not watch</option>
+        <option value="1"{% if issues %} selected{% endif %}>Watch</option>
+      </select>
+    </div>
+    <div class="field">
+      <label for="watch-pull-requests">Pull requests</label>
+      <select id="watch-pull-requests" name="pull-requests">
+        <option value="0"{% if !pull_requests %} selected{% endif %}>Do not watch</option>
+        <option value="1"{% if pull_requests %} selected{% endif %}>Watch</option>
+      </select>
+    </div>
+    <button type="submit">Save watch preferences</button>
+  </form>
+  <p>Set all three values to “Do not watch” to stop watching this repository.</p>
+{% else %}
+  <p><a href="/login">Log in</a> to change watch preferences.</p>
+{% endif %}
+{% endblock %}

@@ -21,7 +21,8 @@
     include_str!("../src/store/migrations/010_audit_history.sql"),
     include_str!("../src/store/migrations/011_domain_events.sql"),
     include_str!("../src/store/migrations/012_issues.sql"),
-    "PRAGMA user_version = 12;\n",
+    include_str!("../src/store/migrations/013_watches.sql"),
+    "PRAGMA user_version = 13;\n",
 );
 
 #[test]

@@ -55,6 +55,9 @@
 )]
 #[path = "../src/store/mod.rs"]
 mod store;
+#[allow(dead_code, reason = "the public-route test does not change watches")]
+#[path = "../src/watch.rs"]
+mod watch;
 
 use std::collections::BTreeMap;
 use std::fs;
@@ -722,6 +725,80 @@
     let feed = request(server.address(), "GET", "/alice/example/atom.xml", &[], &[]);
     assert_eq!(feed.status, 200);
     assert!(feed.text().contains("alice reopened #1"));
+
+    let anonymous_watch = request(server.address(), "GET", "/alice/example/watch", &[], &[]);
+    assert_eq!(anonymous_watch.status, 200);
+    assert!(
+        anonymous_watch
+            .text()
+            .contains("Log in</a> to change watch preferences.")
+    );
+    let watch_page = request(
+        server.address(),
+        "GET",
+        "/alice/example/watch",
+        &[("Cookie", cookie.as_str())],
+        &[],
+    );
+    assert_eq!(watch_page.status, 200);
+    assert!(
+        watch_page
+            .text()
+            .contains("You do not watch this repository.")
+    );
+    let everything = form(&[
+        ("csrf", csrf.as_str()),
+        ("pushes", "1"),
+        ("issues", "1"),
+        ("pull-requests", "1"),
+    ]);
+    let watched = request(
+        server.address(),
+        "POST",
+        "/alice/example/watch",
+        &headers,
+        everything.as_bytes(),
+    );
+    assert_eq!(watched.status, 303);
+    assert_eq!(watched.header("location"), "/alice/example/watch");
+    let selected = request(
+        server.address(),
+        "GET",
+        "/alice/example/watch",
+        &[("Cookie", cookie.as_str())],
+        &[],
+    );
+    assert!(
+        selected
+            .text()
+            .contains("You watch selected activity in this repository.")
+    );
+    assert_eq!(selected.text().matches("value=\"1\" selected").count(), 3);
+    let none = form(&[
+        ("csrf", csrf.as_str()),
+        ("pushes", "0"),
+        ("issues", "0"),
+        ("pull-requests", "0"),
+    ]);
+    assert_eq!(
+        request(
+            server.address(),
+            "POST",
+            "/alice/example/watch",
+            &headers,
+            none.as_bytes(),
+        )
+        .status,
+        303
+    );
+    assert_eq!(
+        Store::open(&database)
+            .expect("reopen the watch database")
+            .connection()
+            .query_row("SELECT count(*) FROM watch", [], |row| row.get::<_, i64>(0))
+            .expect("count cleared watches"),
+        0
+    );
 
     server.shutdown().await.expect("stop the issue Web server");
 }

@@ -11,7 +11,7 @@
 use rusqlite::{Connection, ErrorCode, TransactionBehavior, params};
 use store::{
     GitOperationIntent, InitialAdministrator, IssueChange, NewAuditEvent, NewIssue, NewRepository,
-    NewRepositoryReference, RepositoryOrigin, Store, StoreError,
+    NewRepositoryReference, RepositoryOrigin, Store, StoreError, WatchPreferences,
 };
 use tempfile::TempDir;
 
@@ -47,6 +47,15 @@
     include_str!("../src/store/migrations/010_audit_history.sql"),
     include_str!("../src/store/migrations/011_domain_events.sql"),
     "PRAGMA user_version = 11;\n",
+);
+const V12_FIXTURE: &str = concat!(
+    include_str!("fixtures/sqlite/v7.sql"),
+    include_str!("../src/store/migrations/008_web_sessions.sql"),
+    include_str!("../src/store/migrations/009_repository_authorization.sql"),
+    include_str!("../src/store/migrations/010_audit_history.sql"),
+    include_str!("../src/store/migrations/011_domain_events.sql"),
+    include_str!("../src/store/migrations/012_issues.sql"),
+    "PRAGMA user_version = 12;\n",
 );
 
 fn database(directory: &TempDir, name: &str) -> std::path::PathBuf {
@@ -165,7 +174,7 @@
     let directory = TempDir::new().expect("create a temporary directory");
     let store = Store::open(&database(&directory, "store.sqlite")).expect("open the store");
 
-    assert_eq!(store.schema_version().expect("read the schema version"), 12);
+    assert_eq!(store.schema_version().expect("read the schema version"), 13);
     assert_eq!(
         store
             .connection()
@@ -872,6 +881,96 @@
         assert!(!event.actor.is_empty());
     }
 
+    let events_before_watch: i64 = store
+        .connection()
+        .query_row("SELECT count(*) FROM repository_event", [], |row| {
+            row.get(0)
+        })
+        .expect("count events before watch changes");
+    let everything = WatchPreferences {
+        pushes: true,
+        issues: true,
+        pull_requests: true,
+    };
+    let watch = store
+        .set_watch("alice", "project", "bob", everything, 16)
+        .expect("watch all repository activity")
+        .expect("read the created watch");
+    assert_eq!(watch.id.len(), 32);
+    assert!(watch.pushes && watch.issues && watch.pull_requests);
+    assert_eq!(watch.created_at, 16);
+    assert_eq!(watch.updated_at, 16);
+    let granular = WatchPreferences {
+        pushes: false,
+        issues: true,
+        pull_requests: false,
+    };
+    let updated = store
+        .set_watch("alice", "project", "bob", granular, 17)
+        .expect("set granular watch preferences")
+        .expect("read the updated watch");
+    assert_eq!(updated.id, watch.id);
+    assert_eq!(updated.created_at, 16);
+    assert_eq!(updated.updated_at, 17);
+    assert!(!updated.pushes && updated.issues && !updated.pull_requests);
+    let (watched_repository, stored_watch) = store
+        .watch("alice", "project", Some("bob"))
+        .expect("read watch preferences");
+    assert_eq!(watched_repository.slug, "project");
+    assert_eq!(stored_watch.expect("find the watch"), updated);
+    assert!(matches!(
+        store.watch("alice", "project", Some("stranger")),
+        Err(StoreError::WatchDenied)
+    ));
+    assert!(matches!(
+        store.set_watch("alice", "project", "suspended", everything, 18),
+        Err(StoreError::WatchDenied)
+    ));
+    assert_eq!(
+        store
+            .connection()
+            .query_row("SELECT count(*) FROM repository_event", [], |row| row
+                .get::<_, i64>(0))
+            .expect("count events after watch changes"),
+        events_before_watch
+    );
+    assert_eq!(
+        store
+            .set_watch(
+                "alice",
+                "project",
+                "bob",
+                WatchPreferences {
+                    pushes: false,
+                    issues: false,
+                    pull_requests: false,
+                },
+                19,
+            )
+            .expect("stop watching the repository"),
+        None
+    );
+    assert_eq!(
+        store
+            .watch("alice", "project", Some("bob"))
+            .expect("read the cleared watch")
+            .1,
+        None
+    );
+    assert!(
+        store
+            .connection()
+            .execute(
+                "INSERT INTO watch
+             (id, repository_id, account_id, pushes, issues, pull_requests,
+              created_at, updated_at)
+             VALUES ('aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa',
+                     '00112233445566778899aabbccddeeff', 2, 0, 0, 0, 20, 20)",
+                [],
+            )
+            .is_err()
+    );
+
     let comments_before: i64 = store
         .connection()
         .query_row("SELECT count(*) FROM issue_comment", [], |row| row.get(0))
@@ -887,7 +986,7 @@
         )
         .expect("inject an issue event failure");
     assert!(matches!(
-        store.comment_issue("alice", "project", 1, "bob", "must roll back", 16),
+        store.comment_issue("alice", "project", 1, "bob", "must roll back", 21),
         Err(StoreError::Sqlite(_))
     ));
     assert_eq!(
@@ -1239,13 +1338,14 @@
         (V9_FIXTURE, 9),
         (V10_FIXTURE, 10),
         (V11_FIXTURE, 11),
+        (V12_FIXTURE, 12),
     ] {
         let directory = TempDir::new().expect("create a temporary directory");
         let path = database(&directory, "tit.sqlite3");
         create_fixture(&path, fixture);
 
         let store = Store::open(&path).expect("migrate the fixture");
-        assert_eq!(store.schema_version().expect("read the schema version"), 12);
+        assert_eq!(store.schema_version().expect("read the schema version"), 13);
         store.integrity_check().expect("check migrated integrity");
         let state: String = store
             .connection()
@@ -1318,7 +1418,7 @@
 
 #[test]
 fn recovers_complete_schema_versions_after_a_process_kill_during_migration() {
-    for (mode, expected_version) in [("migration-uncommitted", 1), ("migration-committed", 12)] {
+    for (mode, expected_version) in [("migration-uncommitted", 1), ("migration-committed", 13)] {
         let directory = TempDir::new().expect("create a temporary directory");
         let path = database(&directory, "fixture.sqlite");
         create_fixture(&path, V1_FIXTURE);

@@ -47,6 +47,9 @@
 #[allow(dead_code, reason = "the shell test does not use repository storage")]
 #[path = "../src/store/mod.rs"]
 mod store;
+#[allow(dead_code, reason = "the Web shell test does not change watches")]
+#[path = "../src/watch.rs"]
+mod watch;
 
 use std::collections::BTreeMap;
 use std::io::{Read, Write};