Files
node/crates/gitlawb-node/src/webhooks.rs
T
Vasanth TandClaude Opus 4.8 2ce4da9cc5 feat(node): graceful shutdown + Prometheus metrics endpoint (#22)
* feat(db): versioned schema migrations with idempotent backfill

Replace the inline &str SQL array that re-runs on every node startup with a
versioned migration system. Each migration is recorded in a new
schema_migrations table and applied at most once per node, inside a single
transaction.

For nodes upgrading from a pre-migration-versioning build, the canonical
`repos` table is used as a signal to mark v1 as already applied without
re-running its ~140 statements — so existing operators see zero behavior
change on first restart after this commit.

Closes the gap called out in docs/OSS-READINESS-AUDIT.md:78.

- Adds Db::migration_status() returning applied (version, name, applied_at)
- Adds 6 unit tests validating the static migration catalogue
  (versions strictly increasing, names distinct, bodies non-empty, v1
  name locked to initial_schema)
- cargo fmt + clippy -D warnings + cargo test --workspace all clean

* fix(db): make legacy migration backfill safe and lock-guarded

Address two robustness gaps in the v1 migration runner on the live network:

- Remove the "repos table present => v1 complete" backfill short-circuit.
  It assumed every existing node already had the full current schema; a node
  that was behind would be marked v1-applied while still missing newer tables
  or columns. Every v1 statement is idempotent (CREATE TABLE/INDEX IF NOT
  EXISTS, ADD COLUMN IF NOT EXISTS), so legacy installs now simply run v1 once:
  existing objects are no-ops, missing ones get created.

- Guard the whole runner with a Postgres advisory lock so two instances
  against the same database (blue/green or rolling deploy) can't race to apply
  the same migration and trip the schema_migrations primary key. The lock is
  released explicitly, and automatically if the connection drops.

Also document that per-migration transactions preclude CREATE INDEX
CONCURRENTLY in future migrations.

Co-Authored-By: Claude Opus 4.8 <[email protected]>

* feat(node): graceful shutdown + Prometheus metrics endpoint

Two operational-safety changes that the live network needs:

1) Graceful shutdown. A process-wide tokio::sync::watch::Sender<bool>
   in AppState is flipped on SIGINT (all platforms) or SIGTERM (Unix).
   Every long-lived loop now selects on the shutdown arm:
     * axum server (with_graceful_shutdown — drains in-flight)
     * libp2p swarm (drops Swarm, closes connections)
     * gossip task (exits between peer announces)
     * sync worker (exits between batches)
     * operator heartbeat loop
     * rate-limit cleanup loop
     * peer-count metrics poller
     * the deferred DID-record publish task

   New GITLAWB_SHUTDOWN_GRACE_SECS (default 30s) controls how long
   axum waits for in-flight requests before returning.

2) Prometheus /metrics endpoint. New metrics module exposes:
     * gitlawb_info{version, did}                 (constant gauge = 1)
     * gitlawb_pushes_total{repo}                 (counter)
     * gitlawb_fetches_total{repo}                (counter)
     * gitlawb_auth_successes_total{route}        (counter, helper)
     * gitlawb_auth_failures_total{route,reason}  (counter, helper)
     * gitlawb_sync_queue_processed_total{status} (counter)
     * gitlawb_webhook_deliveries_total{result}   (counter)
     * gitlawb_pack_size_bytes                    (histogram)
     * gitlawb_peers_connected                    (gauge)

   Opt-in via GITLAWB_METRICS_ADDR (e.g. 127.0.0.1:9091). The endpoint
   is on a separate listener so the public port stays unexposed;
   bind to localhost or a private interface.

3) Metric instrumentation hooks in api/repos.rs (push/fetch),
   webhooks.rs (delivery outcomes), and sync.rs (per-item status).
   The auth helpers are defined but not yet wired — that's a
   focused follow-up.

Tests:
  * 3 new metrics::tests::* — encode returns valid prometheus text,
    record_* helpers are no-ops before init(), encode() errors if
    init() was never called.
  * cargo fmt + clippy -D warnings + cargo test --workspace all clean.

Closes the metrics and graceful-shutdown gaps in
docs/OSS-READINESS-AUDIT.md:131 and docs/MAINTAINER-ROADMAP.md:35.

---------

Co-authored-by: Claude Opus 4.8 <[email protected]>
2026-06-02 08:58:05 +08:00

123 lines
3.6 KiB
Rust

//! Outbound webhook delivery.
//!
//! Events fired:
//! pull_request.opened — PR created
//! pull_request.reviewed — review submitted
//! pull_request.merged — PR merged
//! pull_request.closed — PR closed without merging
//! push — branch pushed
//!
//! Payload headers:
//! Content-Type: application/json
//! X-Gitlawb-Event: <event>
//! X-Gitlawb-Delivery: <uuid>
//! X-Gitlawb-Signature-256: sha256=<hmac-sha256-hex> (only if secret set)
use std::sync::Arc;
use hmac::{Hmac, Mac};
use sha2::Sha256;
use crate::db::Db;
type HmacSha256 = Hmac<Sha256>;
/// Compute `sha256=<hex>` HMAC signature for a webhook payload.
fn sign_payload(secret: &str, payload: &[u8]) -> String {
let mut mac =
HmacSha256::new_from_slice(secret.as_bytes()).expect("HMAC accepts any key length");
mac.update(payload);
format!("sha256={}", hex::encode(mac.finalize().into_bytes()))
}
/// Fire webhooks for `event` on `repo_id`. Spawns background tasks — never blocks.
pub fn fire_event(
db: Arc<Db>,
http_client: Arc<reqwest::Client>,
repo_id: &str,
event: &str,
payload: serde_json::Value,
) {
let repo_id = repo_id.to_string();
let event = event.to_string();
tokio::spawn(async move {
fire_event_async(db, http_client, &repo_id, &event, payload).await;
});
}
async fn fire_event_async(
db: Arc<Db>,
http_client: Arc<reqwest::Client>,
repo_id: &str,
event: &str,
payload: serde_json::Value,
) {
let hooks = match db.list_webhooks_for_event(repo_id, event).await {
Ok(h) => h,
Err(e) => {
tracing::warn!(err = %e, "failed to list webhooks for event {event}");
return;
}
};
if hooks.is_empty() {
return;
}
let payload_bytes = match serde_json::to_vec(&payload) {
Ok(b) => b,
Err(e) => {
tracing::warn!(err = %e, "failed to serialize webhook payload");
return;
}
};
for hook in hooks {
let client = Arc::clone(&http_client);
let event_name = event.to_string();
let bytes = payload_bytes.clone();
let delivery_id = uuid::Uuid::new_v4().to_string();
tokio::spawn(async move {
let signature = hook.secret.as_deref().map(|s| sign_payload(s, &bytes));
let mut req = client
.post(&hook.url)
.header("Content-Type", "application/json")
.header("X-Gitlawb-Event", &event_name)
.header("X-Gitlawb-Delivery", &delivery_id)
.body(bytes);
if let Some(sig) = signature {
req = req.header("X-Gitlawb-Signature-256", sig);
}
match req.send().await {
Ok(resp) => {
let result_label = if resp.status().is_success() {
"ok"
} else {
"http_error"
};
crate::metrics::record_webhook_delivery(result_label);
tracing::info!(
url = %hook.url,
event = %event_name,
status = %resp.status(),
"webhook delivered"
)
}
Err(e) => {
crate::metrics::record_webhook_delivery("network_error");
tracing::warn!(
url = %hook.url,
event = %event_name,
err = %e,
"webhook delivery failed"
)
}
}
});
}
}