feature: tap admin

src/http/handle_admin_tap.rs

···
+use anyhow::Result;
+use axum::{Form, response::IntoResponse};
+use axum_template::RenderHtml;
+use minijinja::context as template_context;
+use serde::Deserialize;
+use std::time::Duration;
+
+use crate::{
+    contextual_error,
+    http::{
+        context::{AdminRequestContext, admin_template_context},
+        errors::WebError,
+    },
+    select_template,
+};
+
+/// TAP stats response types
+mod tap_stats {
+    use serde::{Deserialize, Serialize};
+
+    #[derive(Debug, Deserialize, Serialize)]
+    pub struct RepoCount {
+        pub repo_count: i64,
+    }
+
+    #[derive(Debug, Deserialize, Serialize)]
+    pub struct RecordCount {
+        pub record_count: i64,
+    }
+
+    #[derive(Debug, Deserialize, Serialize)]
+    pub struct OutboxBuffer {
+        pub outbox_buffer: i64,
+    }
+
+    #[derive(Debug, Deserialize, Serialize)]
+    pub struct ResyncBuffer {
+        pub resync_buffer: i64,
+    }
+
+    #[derive(Debug, Deserialize, Serialize)]
+    pub struct Cursors {
+        pub firehose: i64,
+        pub list_repos: Option<String>,
+    }
+
+    #[derive(Debug, Deserialize, Serialize)]
+    pub struct RepoInfo {
+        pub did: String,
+        pub error: Option<String>,
+        pub handle: Option<String>,
+        pub records: i64,
+        pub retries: i64,
+        pub rev: Option<String>,
+        pub state: String,
+    }
+}
+
+#[derive(Debug, Deserialize)]
+pub(crate) struct TapSubmitForm {
+    subject: String,
+}
+
+#[derive(Debug, Deserialize)]
+pub(crate) struct TapInfoForm {
+    subject: String,
+}
+
+/// GET /admin/tap - TAP instance stats
+pub(crate) async fn handle_admin_tap(
+    admin_ctx: AdminRequestContext,
+) -> Result<impl IntoResponse, WebError> {
+    let canonical_url = format!(
+        "https://{}/admin/tap",
+        admin_ctx.web_context.config.external_base
+    );
+    let default_context = admin_template_context(&admin_ctx, &canonical_url, "tap");
+
+    let render_template = select_template!("admin_tap", false, false, admin_ctx.language);
+
+    let tap_enabled = admin_ctx.web_context.config.enable_tap;
+    let tap_hostname = &admin_ctx.web_context.config.tap_hostname;
+
+    // If TAP is not enabled, show a simple message
+    if !tap_enabled {
+        return Ok(RenderHtml(
+            &render_template,
+            admin_ctx.web_context.engine.clone(),
+            template_context! { ..default_context, ..template_context! {
+                tap_enabled => false,
+                tap_hostname => tap_hostname,
+            }},
+        )
+        .into_response());
+    }
+
+    let base_url = format!("http://{}", tap_hostname);
+    let client = &admin_ctx.web_context.http_client;
+
+    // Fetch all stats endpoints concurrently
+    let (repo_count, record_count, outbox_buffer, resync_buffer, cursors) = tokio::join!(
+        fetch_tap_stat::<tap_stats::RepoCount>(client, &base_url, "/stats/repo-count"),
+        fetch_tap_stat::<tap_stats::RecordCount>(client, &base_url, "/stats/record-count"),
+        fetch_tap_stat::<tap_stats::OutboxBuffer>(client, &base_url, "/stats/outbox-buffer"),
+        fetch_tap_stat::<tap_stats::ResyncBuffer>(client, &base_url, "/stats/resync-buffer"),
+        fetch_tap_stat::<tap_stats::Cursors>(client, &base_url, "/stats/cursors"),
+    );
+
+    // Determine connection status
+    let connected = repo_count.is_some();
+
+    Ok(RenderHtml(
+        &render_template,
+        admin_ctx.web_context.engine.clone(),
+        template_context! { ..default_context, ..template_context! {
+            tap_enabled => true,
+            tap_hostname => tap_hostname,
+            connected => connected,
+            repo_count => repo_count.map(|r| r.repo_count),
+            record_count => record_count.map(|r| r.record_count),
+            outbox_buffer => outbox_buffer.map(|r| r.outbox_buffer),
+            resync_buffer => resync_buffer.map(|r| r.resync_buffer),
+            firehose_cursor => cursors.as_ref().map(|c| c.firehose),
+            list_repos_cursor => cursors.as_ref().and_then(|c| c.list_repos.clone()),
+        }},
+    )
+    .into_response())
+}
+
+/// Helper to fetch a TAP stats endpoint
+async fn fetch_tap_stat<T: serde::de::DeserializeOwned>(
+    client: &reqwest::Client,
+    base_url: &str,
+    path: &str,
+) -> Option<T> {
+    let url = format!("{}{}", base_url, path);
+    match client.get(&url).timeout(Duration::from_secs(30)).send().await {
+        Ok(response) if response.status().is_success() => response.json::<T>().await.ok(),
+        Ok(response) => {
+            tracing::debug!(url = %url, status = %response.status(), "TAP stats request failed");
+            None
+        }
+        Err(e) => {
+            tracing::debug!(url = %url, error = %e, "TAP stats request error");
+            None
+        }
+    }
+}
+
+/// POST /admin/tap/submit - Submit a DID to TAP for crawling
+pub(crate) async fn handle_admin_tap_submit(
+    admin_ctx: AdminRequestContext,
+    Form(form): Form<TapSubmitForm>,
+) -> Result<impl IntoResponse, WebError> {
+    let canonical_url = format!(
+        "https://{}/admin/tap",
+        admin_ctx.web_context.config.external_base
+    );
+    let default_context = admin_template_context(&admin_ctx, &canonical_url, "tap");
+
+    let render_template = select_template!("admin_tap", false, false, admin_ctx.language);
+    let error_template = select_template!(false, false, admin_ctx.language);
+
+    let tap_enabled = admin_ctx.web_context.config.enable_tap;
+    let tap_hostname = &admin_ctx.web_context.config.tap_hostname;
+
+    if !tap_enabled {
+        return contextual_error!(
+            admin_ctx.web_context,
+            admin_ctx.language,
+            error_template,
+            default_context,
+            "TAP is not enabled"
+        );
+    }
+
+    let subject = form.subject.trim();
+
+    // Resolve the subject (handle or DID) to a DID
+    let document = match admin_ctx.web_context.identity_resolver.resolve(subject).await {
+        Ok(doc) => doc,
+        Err(err) => {
+            return contextual_error!(
+                admin_ctx.web_context,
+                admin_ctx.language,
+                error_template,
+                default_context,
+                err
+            );
+        }
+    };
+
+    let did = &document.id;
+
+    let url = format!("http://{}/repos/add", tap_hostname);
+
+    // Build the request payload
+    let payload = serde_json::json!({
+        "dids": [did]
+    });
+
+    // Send the request to TAP
+    let response = match admin_ctx
+        .web_context
+        .http_client
+        .post(&url)
+        .json(&payload)
+        .timeout(Duration::from_secs(10))
+        .send()
+        .await
+    {
+        Ok(resp) => resp,
+        Err(err) => {
+            tracing::error!(error = %err, "TAP request failed");
+            return contextual_error!(
+                admin_ctx.web_context,
+                admin_ctx.language,
+                error_template,
+                default_context,
+                format!("TAP request failed: {}", err)
+            );
+        }
+    };
+
+    if !response.status().is_success() {
+        let status = response.status();
+        let body = response.text().await.unwrap_or_default();
+        tracing::warn!(subject = %subject, did = %did, status = %status, body = %body, "TAP submit failed");
+        return contextual_error!(
+            admin_ctx.web_context,
+            admin_ctx.language,
+            error_template,
+            default_context,
+            format!("TAP returned error: {} - {}", status, body)
+        );
+    }
+
+    tracing::info!(subject = %subject, did = %did, "Submitted DID to TAP for crawling");
+
+    // Re-fetch stats and render the page with a success message
+    let base_url = format!("http://{}", tap_hostname);
+    let client = &admin_ctx.web_context.http_client;
+
+    let (repo_count, record_count, outbox_buffer, resync_buffer, cursors) = tokio::join!(
+        fetch_tap_stat::<tap_stats::RepoCount>(client, &base_url, "/stats/repo-count"),
+        fetch_tap_stat::<tap_stats::RecordCount>(client, &base_url, "/stats/record-count"),
+        fetch_tap_stat::<tap_stats::OutboxBuffer>(client, &base_url, "/stats/outbox-buffer"),
+        fetch_tap_stat::<tap_stats::ResyncBuffer>(client, &base_url, "/stats/resync-buffer"),
+        fetch_tap_stat::<tap_stats::Cursors>(client, &base_url, "/stats/cursors"),
+    );
+
+    let connected = repo_count.is_some();
+
+    Ok(RenderHtml(
+        &render_template,
+        admin_ctx.web_context.engine.clone(),
+        template_context! { ..default_context, ..template_context! {
+            tap_enabled => true,
+            tap_hostname => tap_hostname,
+            connected => connected,
+            repo_count => repo_count.map(|r| r.repo_count),
+            record_count => record_count.map(|r| r.record_count),
+            outbox_buffer => outbox_buffer.map(|r| r.outbox_buffer),
+            resync_buffer => resync_buffer.map(|r| r.resync_buffer),
+            firehose_cursor => cursors.as_ref().map(|c| c.firehose),
+            list_repos_cursor => cursors.as_ref().and_then(|c| c.list_repos.clone()),
+            submit_success => true,
+            submit_subject => subject,
+            submit_did => did.clone(),
+        }},
+    )
+    .into_response())
+}
+
+/// POST /admin/tap/info - Get info about a DID from TAP
+pub(crate) async fn handle_admin_tap_info(
+    admin_ctx: AdminRequestContext,
+    Form(form): Form<TapInfoForm>,
+) -> Result<impl IntoResponse, WebError> {
+    let canonical_url = format!(
+        "https://{}/admin/tap",
+        admin_ctx.web_context.config.external_base
+    );
+    let default_context = admin_template_context(&admin_ctx, &canonical_url, "tap");
+
+    let render_template = select_template!("admin_tap", false, false, admin_ctx.language);
+    let error_template = select_template!(false, false, admin_ctx.language);
+
+    let tap_enabled = admin_ctx.web_context.config.enable_tap;
+    let tap_hostname = &admin_ctx.web_context.config.tap_hostname;
+
+    if !tap_enabled {
+        return contextual_error!(
+            admin_ctx.web_context,
+            admin_ctx.language,
+            error_template,
+            default_context,
+            "TAP is not enabled"
+        );
+    }
+
+    let subject = form.subject.trim();
+
+    // Resolve the subject (handle or DID) to a DID
+    let document = match admin_ctx.web_context.identity_resolver.resolve(subject).await {
+        Ok(doc) => doc,
+        Err(err) => {
+            return contextual_error!(
+                admin_ctx.web_context,
+                admin_ctx.language,
+                error_template,
+                default_context,
+                err
+            );
+        }
+    };
+
+    let did = &document.id;
+
+    let url = format!("http://{}/info/{}", tap_hostname, did);
+
+    // Send the request to TAP
+    let response = match admin_ctx
+        .web_context
+        .http_client
+        .get(&url)
+        .timeout(Duration::from_secs(10))
+        .send()
+        .await
+    {
+        Ok(resp) => resp,
+        Err(err) => {
+            tracing::error!(error = %err, "TAP info request failed");
+            return contextual_error!(
+                admin_ctx.web_context,
+                admin_ctx.language,
+                error_template,
+                default_context,
+                format!("TAP request failed: {}", err)
+            );
+        }
+    };
+
+    let info: Option<tap_stats::RepoInfo> = if response.status().is_success() {
+        response.json().await.ok()
+    } else {
+        None
+    };
+
+    // Re-fetch the TAP stats for the page
+    let base_url = format!("http://{}", tap_hostname);
+    let client = &admin_ctx.web_context.http_client;
+
+    let (repo_count, record_count, outbox_buffer, resync_buffer, cursors) = tokio::join!(
+        fetch_tap_stat::<tap_stats::RepoCount>(client, &base_url, "/stats/repo-count"),
+        fetch_tap_stat::<tap_stats::RecordCount>(client, &base_url, "/stats/record-count"),
+        fetch_tap_stat::<tap_stats::OutboxBuffer>(client, &base_url, "/stats/outbox-buffer"),
+        fetch_tap_stat::<tap_stats::ResyncBuffer>(client, &base_url, "/stats/resync-buffer"),
+        fetch_tap_stat::<tap_stats::Cursors>(client, &base_url, "/stats/cursors"),
+    );
+
+    let connected = repo_count.is_some();
+
+    let info_context = info.as_ref().map(|i| {
+        template_context! {
+            did => i.did.clone(),
+            error => i.error.clone(),
+            handle => i.handle.clone(),
+            records => i.records,
+            retries => i.retries,
+            rev => i.rev.clone(),
+            state => i.state.clone(),
+        }
+    });
+
+    Ok(RenderHtml(
+        &render_template,
+        admin_ctx.web_context.engine.clone(),
+        template_context! { ..default_context, ..template_context! {
+            tap_enabled => true,
+            tap_hostname => tap_hostname,
+            connected => connected,
+            repo_count => repo_count.map(|r| r.repo_count),
+            record_count => record_count.map(|r| r.record_count),
+            outbox_buffer => outbox_buffer.map(|r| r.outbox_buffer),
+            resync_buffer => resync_buffer.map(|r| r.resync_buffer),
+            firehose_cursor => cursors.as_ref().map(|c| c.firehose),
+            list_repos_cursor => cursors.as_ref().and_then(|c| c.list_repos.clone()),
+            info_subject => subject,
+            info_result => info_context,
+            info_not_found => info.is_none(),
+        }},
+    )
+    .into_response())
+}

src/http/mod.rs

···
 pub mod handle_admin_rsvp_accepts;
 pub mod handle_admin_rsvps;
 pub mod handle_admin_search_index;
+pub mod handle_admin_tap;
 pub mod handle_api_mcp_configuration;
 pub mod handle_blob;
 pub mod handle_bulk_accept_rsvps;

src/http/server.rs

···
         handle_admin_search_index, handle_admin_search_index_delete,
         handle_admin_search_index_rebuild,
     },
+    handle_admin_tap::{handle_admin_tap, handle_admin_tap_info, handle_admin_tap_submit},
     handle_api_mcp_configuration::{
         handle_api_mcp_configuration_get, handle_api_mcp_configuration_post,
     },
···
             "/admin/search-index/profiles/rebuild",
             post(handle_admin_profile_index_rebuild),
         )
+        .route("/admin/tap", get(handle_admin_tap))
+        .route("/admin/tap/submit", post(handle_admin_tap_submit))
+        .route("/admin/tap/info", post(handle_admin_tap_info))
         .route("/content/{cid}", get(handle_content))
         .route("/logout", get(handle_auth_logout))
         .route("/language", post(handle_set_language))

templates/en-us/admin_base.html

···
                         <li><a href="/admin/search-index/events" {% if active_section == "search-index-events" %}class="is-active"{% endif %}>Event Index</a></li>
                         <li><a href="/admin/search-index/profiles" {% if active_section == "search-index-profiles" %}class="is-active"{% endif %}>Profile Index</a></li>
                     </ul>
+                    <p class="menu-label">
+                        Infrastructure
+                    </p>
+                    <ul class="menu-list">
+                        <li><a href="/admin/tap" {% if active_section == "tap" %}class="is-active"{% endif %}>TAP</a></li>
+                    </ul>
                 </aside>
             </div>
             <div class="column">

templates/en-us/admin_tap.html

···
+{% extends "en-us/admin_base.html" %}
+{% block admin_title %}TAP Management{% endblock %}
+{% block admin_content %}
+<div class="content">
+    <h1 class="title">TAP Management</h1>
+    <p class="subtitle">Monitor and manage the TAP (Timeline Aggregation Protocol) instance</p>
+
+    {% if not tap_enabled %}
+    <article class="message is-warning">
+        <div class="message-header">
+            <p>TAP Disabled</p>
+        </div>
+        <div class="message-body">
+            TAP is not enabled in the current configuration. Set <code>ENABLE_TAP=true</code> to enable TAP functionality.
+            <br><br>
+            <strong>Configured hostname:</strong> <code>{{ tap_hostname }}</code>
+        </div>
+    </article>
+    {% else %}
+
+    {% if submit_success %}
+    <article class="message is-success">
+        <div class="message-body">
+            Successfully submitted <strong>{{ submit_subject }}</strong> (<code>{{ submit_did }}</code>) to TAP for crawling.
+        </div>
+    </article>
+    {% endif %}
+
+    <div class="box mb-5">
+        <h2 class="title is-4">Connection Status</h2>
+        <div class="columns">
+            <div class="column">
+                <p>
+                    <strong>Hostname:</strong> <code>{{ tap_hostname }}</code>
+                </p>
+                <p>
+                    <strong>Status:</strong>
+                    {% if connected %}
+                    <span class="tag is-success">Connected</span>
+                    {% else %}
+                    <span class="tag is-danger">Disconnected</span>
+                    {% endif %}
+                </p>
+            </div>
+        </div>
+    </div>
+
+    {% if connected %}
+    <div class="box mb-5">
+        <h2 class="title is-4">Instance Statistics</h2>
+        <div class="columns is-multiline">
+            <div class="column is-one-quarter">
+                <div class="notification is-light">
+                    <p class="heading">Repositories</p>
+                    <p class="title is-4">{{ repo_count | default("-") }}</p>
+                </div>
+            </div>
+            <div class="column is-one-quarter">
+                <div class="notification is-light">
+                    <p class="heading">Records</p>
+                    <p class="title is-4">{{ record_count | default("-") }}</p>
+                </div>
+            </div>
+            <div class="column is-one-quarter">
+                <div class="notification is-light">
+                    <p class="heading">Outbox Buffer</p>
+                    <p class="title is-4">{{ outbox_buffer | default("-") }}</p>
+                </div>
+            </div>
+            <div class="column is-one-quarter">
+                <div class="notification is-light">
+                    <p class="heading">Resync Buffer</p>
+                    <p class="title is-4">{{ resync_buffer | default("-") }}</p>
+                </div>
+            </div>
+        </div>
+        <div class="columns">
+            <div class="column">
+                <p><strong>Firehose Cursor:</strong> <code>{{ firehose_cursor | default("N/A") }}</code></p>
+                <p><strong>List Repos Cursor:</strong> <code>{{ list_repos_cursor | default("N/A") }}</code></p>
+            </div>
+        </div>
+    </div>
+    {% endif %}
+
+    <div class="columns">
+        <div class="column is-half">
+            <div class="box">
+                <h2 class="title is-4">Submit DID for Crawling</h2>
+                <form action="/admin/tap/submit" method="POST">
+                    <div class="field">
+                        <label class="label" for="submitSubjectInput">Handle or DID</label>
+                        <div class="control has-icons-left">
+                            <input class="input" type="text" id="submitSubjectInput" placeholder="handle.bsky.social or did:plc:..." name="subject" required="required">
+                            <span class="icon is-small is-left">
+                                <i class="fas fa-user"></i>
+                            </span>
+                        </div>
+                        <p class="help">Enter a handle or DID to submit for crawling</p>
+                    </div>
+                    <div class="field">
+                        <div class="control">
+                            <button type="submit" class="button is-primary">Submit</button>
+                        </div>
+                    </div>
+                </form>
+            </div>
+        </div>
+
+        <div class="column is-half">
+            <div class="box">
+                <h2 class="title is-4">Lookup Repository Info</h2>
+                <form action="/admin/tap/info" method="POST">
+                    <div class="field">
+                        <label class="label" for="infoSubjectInput">Handle or DID</label>
+                        <div class="control has-icons-left">
+                            <input class="input" type="text" id="infoSubjectInput" placeholder="handle.bsky.social or did:plc:..." name="subject" required="required" {% if info_subject %}value="{{ info_subject }}"{% endif %}>
+                            <span class="icon is-small is-left">
+                                <i class="fas fa-search"></i>
+                            </span>
+                        </div>
+                        <p class="help">Enter a handle or DID to lookup repository info</p>
+                    </div>
+                    <div class="field">
+                        <div class="control">
+                            <button type="submit" class="button is-info">Lookup</button>
+                        </div>
+                    </div>
+                </form>
+            </div>
+        </div>
+    </div>
+
+    {% if info_subject %}
+    <div class="box">
+        <h2 class="title is-4">Repository Info: {{ info_subject }}</h2>
+        {% if info_not_found %}
+        <article class="message is-warning">
+            <div class="message-body">
+                No repository found for <strong>{{ info_subject }}</strong>. The DID may not be tracked by this TAP instance.
+            </div>
+        </article>
+        {% elif info_result %}
+        <table class="table is-fullwidth">
+            <tbody>
+                <tr>
+                    <th style="width: 150px;">DID</th>
+                    <td><code>{{ info_result.did }}</code></td>
+                </tr>
+                {% if info_result.handle %}
+                <tr>
+                    <th>Handle</th>
+                    <td>{{ info_result.handle }}</td>
+                </tr>
+                {% endif %}
+                <tr>
+                    <th>State</th>
+                    <td>
+                        {% if info_result.state == "synced" %}
+                        <span class="tag is-success">{{ info_result.state }}</span>
+                        {% elif info_result.state == "syncing" %}
+                        <span class="tag is-info">{{ info_result.state }}</span>
+                        {% elif info_result.state == "pending" %}
+                        <span class="tag is-warning">{{ info_result.state }}</span>
+                        {% elif info_result.state == "failed" %}
+                        <span class="tag is-danger">{{ info_result.state }}</span>
+                        {% else %}
+                        <span class="tag">{{ info_result.state }}</span>
+                        {% endif %}
+                    </td>
+                </tr>
+                <tr>
+                    <th>Records</th>
+                    <td>{{ info_result.records }}</td>
+                </tr>
+                <tr>
+                    <th>Retries</th>
+                    <td>{{ info_result.retries }}</td>
+                </tr>
+                {% if info_result.rev %}
+                <tr>
+                    <th>Revision</th>
+                    <td><code class="is-size-7">{{ info_result.rev }}</code></td>
+                </tr>
+                {% endif %}
+                {% if info_result.error %}
+                <tr>
+                    <th>Error</th>
+                    <td><span class="has-text-danger">{{ info_result.error }}</span></td>
+                </tr>
+                {% endif %}
+            </tbody>
+        </table>
+        {% endif %}
+    </div>
+    {% endif %}
+
+    {% endif %}
+</div>
+{% endblock %}