feat: route trigger callback room and enforce peer/issue sync policies

Author: mizchi

README.md

@@ -175,9 +175,11 @@ optional env:
 - `RELAY_NODE_ID` (cache exchange ノード ID)
 - `RELAY_PEERS` (CSV: peer relay URLs)
 - `RELAY_PEERS_JSON` (JSON: peer relay URLs, `RELAY_PEERS` より優先)
-- `RELAY_PEER_AUTH_TOKEN` (peer pull 時の Bearer token)
+- `RELAY_PEER_AUTH_TOKEN` (relay 間共有 Bearer token。設定時は `/api/v1/cache/exchange/*` と
+  `/api/v1/cache/issues/*` が必須認証)
 - `RELAY_PEER_SYNC_INTERVAL_SEC` (default: `30`)
 - `RELAY_PEER_REPO_FF_COMMIT_WINDOW` (default: `30`, peer 間 repo 判定で見る commit 数)
+- `RELAY_ISSUE_SOURCE_OF_TRUTH` (`last_write` | `github` | `bit`, default: `last_write`)
 - `RELAY_REPO_ID` (明示 repo ID, 例: `bit-vcs/bit`)
 - `RELAY_REPO_ORIGIN_URL` (origin URL の明示上書き)
 - `RELAY_REPO_RECENT_COMMITS` (CSV: 最近 commit hash)

docs/host-bit-relay.md

@@ -66,9 +66,10 @@ All configuration is done via environment variables (set in the Cloudflare dashb
 
 ### Authentication
 
-| Variable               | Description                                                                                                                           | Default      |
-| ---------------------- | ------------------------------------------------------------------------------------------------------------------------------------- | ------------ |
-| `BIT_RELAY_AUTH_TOKEN` | Bearer token for API authentication. When set, all `/api/v1/*` and `/ws` requests require this token. Leave unset for a public relay. | (none, open) |
+| Variable                | Description                                                                                                                           | Default      |
+| ----------------------- | ------------------------------------------------------------------------------------------------------------------------------------- | ------------ |
+| `BIT_RELAY_AUTH_TOKEN`  | Bearer token for API authentication. When set, all `/api/v1/*` and `/ws` requests require this token. Leave unset for a public relay. | (none, open) |
+| `RELAY_PEER_AUTH_TOKEN` | Shared bearer token for relay-to-relay cache endpoints (`/api/v1/cache/exchange/*`, `/api/v1/cache/issues/*`).                        | (none)       |
 
 ### Signature Verification
 
@@ -91,12 +92,13 @@ All configuration is done via environment variables (set in the Cloudflare dashb
 
 ### Rooms and Sessions
 
-| Variable                      | Description                              | Default            |
-| ----------------------------- | ---------------------------------------- | ------------------ |
-| `RELAY_MAX_MESSAGES_PER_ROOM` | Max stored messages per room             | (built-in default) |
-| `RELAY_ROOM_TOKENS`           | JSON object mapping room names to tokens | `{}`               |
-| `RELAY_PRESENCE_TTL_SEC`      | Presence heartbeat TTL                   | (built-in default) |
-| `GIT_SERVE_SESSION_TTL_SEC`   | Git serve session TTL (0 = no expiry)    | `0`                |
+| Variable                      | Description                                                    | Default            |
+| ----------------------------- | -------------------------------------------------------------- | ------------------ |
+| `RELAY_MAX_MESSAGES_PER_ROOM` | Max stored messages per room                                   | (built-in default) |
+| `RELAY_ROOM_TOKENS`           | JSON object mapping room names to tokens                       | `{}`               |
+| `RELAY_PRESENCE_TTL_SEC`      | Presence heartbeat TTL                                         | (built-in default) |
+| `GIT_SERVE_SESSION_TTL_SEC`   | Git serve session TTL (0 = no expiry)                          | `0`                |
+| `RELAY_ISSUE_SOURCE_OF_TRUTH` | Issue snapshot conflict policy (`last_write`, `github`, `bit`) | `last_write`       |
 
 ### WebSocket
 
@@ -119,6 +121,10 @@ wrangler secret put RELAY_REQUIRE_SIGNATURE
 # Set a room token
 wrangler secret put RELAY_ROOM_TOKENS
 # Enter: {"my-room":"secret-token"}
+
+# Set relay-to-relay shared token for cache exchange
+wrangler secret put RELAY_PEER_AUTH_TOKEN
+# Enter: <shared-token>
 ```
 
 ## Verify Deployment

src/cloudflare_worker.ts

@@ -46,6 +46,7 @@ export interface RelayWorkerEnv {
   RELAY_MAX_CLOCK_SKEW_SEC?: string;
   RELAY_NONCE_TTL_SEC?: string;
   RELAY_MAX_NONCES_PER_SENDER?: string;
+  RELAY_PEER_AUTH_TOKEN?: string;
   RELAY_PRESENCE_TTL_SEC?: string;
   RELAY_IP_PUBLISH_LIMIT_PER_WINDOW?: string;
   RELAY_ROOM_PUBLISH_LIMIT_PER_WINDOW?: string;
@@ -59,6 +60,7 @@ export interface RelayWorkerEnv {
 }
 
 const IDENTITY_KEY = 'relay_identity_v1';
+const INCOMING_TRIGGER_REF_PREFIX = 'refs/relay/incoming/';
 let fallbackService: MemoryRelayService | null = null;
 let adminGitHubApi: ReturnType<typeof createAdminGitHubApi> | null = null;
 const requestMetrics = createRelayRequestMetricRecorder();
@@ -289,6 +291,48 @@ function invalidRoomResponse(): Response {
   return Response.json({ ok: false, error: 'invalid room' }, { status: 400 });
 }
 
+function isObjectRecord(value: unknown): value is Record<string, unknown> {
+  return typeof value === 'object' && value !== null && !Array.isArray(value);
+}
+
+function deriveRoomFromIncomingRef(ref: string): string {
+  const trimmed = ref.trim();
+  if (!trimmed.startsWith(INCOMING_TRIGGER_REF_PREFIX)) return DEFAULT_ROOM;
+  const suffix = trimmed.slice(INCOMING_TRIGGER_REF_PREFIX.length).trim();
+  if (suffix.length === 0) return DEFAULT_ROOM;
+  const first = suffix.split('/')[0].trim();
+  if (!isValidRoomName(first)) return DEFAULT_ROOM;
+  return first;
+}
+
+async function resolveRelayRouteRoom(url: URL, request: Request): Promise<string> {
+  const roomFromQuery = (url.searchParams.get('room') ?? '').trim();
+  if (roomFromQuery.length > 0) {
+    return roomFromQuery;
+  }
+
+  if (url.pathname === '/api/v1/trigger/callback' && request.method === 'POST') {
+    try {
+      const bodyText = await request.clone().text();
+      const parsed = JSON.parse(bodyText);
+      if (isObjectRecord(parsed)) {
+        const roomFromBody = (typeof parsed.room === 'string' ? parsed.room : '').trim();
+        if (roomFromBody.length > 0) {
+          return roomFromBody;
+        }
+        const ref = (typeof parsed.ref === 'string' ? parsed.ref : '').trim();
+        if (ref.length > 0) {
+          return deriveRoomFromIncomingRef(ref);
+        }
+      }
+    } catch {
+      // keep default room fallback
+    }
+  }
+
+  return DEFAULT_ROOM;
+}
+
 export class RelayRoom {
   private readonly state: DurableObjectStateLike;
   private readonly service: MemoryRelayService;
@@ -393,7 +437,7 @@ async function handleWorkerRequest(request: Request, env: RelayWorkerEnv): Promi
     return Response.json({ ok: false, error: 'not found' }, { status: 404 });
   }
 
-  const room = (url.searchParams.get('room') ?? DEFAULT_ROOM).trim();
+  const room = (await resolveRelayRouteRoom(url, request)).trim();
   if (!isValidRoomName(room)) {
     return invalidRoomResponse();
   }

src/deno_main.ts

@@ -79,6 +79,7 @@ const relayCacheStore = createMemoryCacheStore({
 });
 const service = createMemoryRelayService({
   ...runtimeConfig.relay,
+  peerAuthToken: runtimeConfig.peers.authToken ?? runtimeConfig.relay.peerAuthToken,
   peerRelayUrls: runtimeConfig.peers.urls.length > 0
     ? runtimeConfig.peers.urls
     : runtimeConfig.relay.peerRelayUrls,

src/memory_handler.ts

@@ -20,6 +20,7 @@ import { createRelayCacheAdapter } from './relay_cache_adapter.ts';
 
 export type JsonPrimitive = string | number | boolean | null;
 export type JsonValue = JsonPrimitive | JsonValue[] | { [key: string]: JsonValue };
+export type IssueSourceOfTruth = 'last_write' | 'github' | 'bit';
 
 type JsonObject = { [key: string]: JsonValue };
 type KeyStatus = 'active' | 'revoked';
@@ -180,6 +181,8 @@ export interface IdentitySnapshot {
 
 export interface MemoryRelayOptions {
   authToken?: string;
+  peerAuthToken?: string;
+  issueSourceOfTruth?: IssueSourceOfTruth;
   maxMessagesPerRoom?: number;
   publishPayloadMaxBytes?: number;
   publishLimitPerWindow?: number;
@@ -237,6 +240,7 @@ const DEFAULT_WS_PING_INTERVAL_MS = 30_000;
 const DEFAULT_WS_IDLE_TIMEOUT_MS = 90_000;
 const DEFAULT_CACHE_EXCHANGE_MAX_HOPS = 3;
 const DEFAULT_CACHE_EXCHANGE_MAX_RECORDS = 10_000;
+const DEFAULT_ISSUE_SOURCE_OF_TRUTH: IssueSourceOfTruth = 'last_write';
 const MAX_GITHUB_WEBHOOK_DELIVERY_IDS = 10_000;
 const MAX_GITHUB_WEBHOOK_DLQ_ENTRIES = 10_000;
 const GITHUB_WEBHOOK_RETRY_BASE_SEC = 30;
@@ -1100,6 +1104,7 @@ function parseBoolOrDefault(raw: boolean | undefined, fallback: boolean): boolea
 
 export function createMemoryRelayService(options: MemoryRelayOptions = {}): MemoryRelayService {
   const authToken = normalizeAuthToken(options.authToken);
+  const peerAuthToken = normalizeAuthToken(options.peerAuthToken);
   const roomTokens = parseRoomTokens(options.roomTokens);
   const maxMessagesPerRoom = Math.max(
     1,
@@ -1165,6 +1170,10 @@ export function createMemoryRelayService(options: MemoryRelayOptions = {}): Memo
   const cacheStore = options.cacheStore ?? null;
   const githubWebhookSecret = (options.githubWebhookSecret ?? '').trim();
   const fetchFn = options.fetchFn ?? globalThis.fetch;
+  const issueSourceOfTruth = options.issueSourceOfTruth === 'github' ||
+      options.issueSourceOfTruth === 'bit'
+    ? options.issueSourceOfTruth
+    : DEFAULT_ISSUE_SOURCE_OF_TRUTH;
   const cacheAdapter = createRelayCacheAdapter({
     cacheStore,
     isValidRoomName,
@@ -1173,6 +1182,7 @@ export function createMemoryRelayService(options: MemoryRelayOptions = {}): Memo
     cachePersistMaxRetries: options.cachePersistMaxRetries,
     cachePersistRetryBaseDelayMs: options.cachePersistRetryBaseDelayMs,
     cachePersistRetryMaxDelayMs: options.cachePersistRetryMaxDelayMs,
+    issueSourceOfTruth,
   });
 
   const rooms = new Map<string, RoomState>();
@@ -1386,10 +1396,21 @@ export function createMemoryRelayService(options: MemoryRelayOptions = {}): Memo
     return null;
   }
 
+  function checkPeerAuth(request: Request): Response | null {
+    if (peerAuthToken.length === 0) return null;
+    const presented = extractPresentedToken(request);
+    if (presented.length === 0 || !timingSafeEqual(presented, peerAuthToken)) {
+      return unauthorizedResponse();
+    }
+    return null;
+  }
+
   function handleCacheExchangeDiscovery(request: Request): Response {
     if (request.method !== 'GET') {
       return methodNotAllowedResponse();
     }
+    const peerAuthError = checkPeerAuth(request);
+    if (peerAuthError) return peerAuthError;
     return Response.json({
       ok: true,
       protocol: 'cache.exchange.v1',
@@ -1409,6 +1430,8 @@ export function createMemoryRelayService(options: MemoryRelayOptions = {}): Memo
     if (request.method !== 'GET') {
       return methodNotAllowedResponse();
     }
+    const peerAuthError = checkPeerAuth(request);
+    if (peerAuthError) return peerAuthError;
     const after = normalizeAfter(url.searchParams.get('after'), 0);
     const limit = normalizeLimit(url.searchParams.get('limit'), 100);
     const peerRaw = (url.searchParams.get('peer') ?? '').trim();
@@ -1447,6 +1470,8 @@ export function createMemoryRelayService(options: MemoryRelayOptions = {}): Memo
     if (request.method !== 'POST') {
       return methodNotAllowedResponse();
     }
+    const peerAuthError = checkPeerAuth(request);
+    if (peerAuthError) return peerAuthError;
 
     let parsed: unknown;
     try {
@@ -1569,6 +1594,8 @@ export function createMemoryRelayService(options: MemoryRelayOptions = {}): Memo
     if (request.method !== 'GET') {
       return methodNotAllowedResponse();
     }
+    const peerAuthError = checkPeerAuth(request);
+    if (peerAuthError) return peerAuthError;
     const room = normalizeRoom(url.searchParams.get('room'));
     if (!isValidRoomName(room)) {
       return invalidRoomResponse();
@@ -1607,6 +1634,8 @@ export function createMemoryRelayService(options: MemoryRelayOptions = {}): Memo
     if (request.method !== 'GET') {
       return methodNotAllowedResponse();
     }
+    const peerAuthError = checkPeerAuth(request);
+    if (peerAuthError) return peerAuthError;
     const room = normalizeRoom(url.searchParams.get('room'));
     if (!isValidRoomName(room)) {
       return invalidRoomResponse();

src/relay_cache_adapter.ts

@@ -1,4 +1,5 @@
 import type { CacheStore, CacheStoreListResult, CacheStoreObject } from './cache_store.ts';
+import type { IssueSourceOfTruth } from './memory_handler.ts';
 import { createCachePersistenceQueue } from './cache_persistence_queue.ts';
 import { canonicalizeJson, sha256Hex } from './signing.ts';
 import {
@@ -42,6 +43,7 @@ export interface RelayCacheAdapterOptions {
   cacheStore: CacheStore | null;
   isValidRoomName: (room: string) => boolean;
   isValidTopic: (topic: string) => boolean;
+  issueSourceOfTruth?: IssueSourceOfTruth;
   nowSec?: () => number;
   cachePersistMaxRetries?: number;
   cachePersistRetryBaseDelayMs?: number;
@@ -76,6 +78,9 @@ export interface RelayCacheAdapter {
 const DEFAULT_CACHE_PERSIST_MAX_RETRIES = 2;
 const DEFAULT_CACHE_PERSIST_RETRY_BASE_DELAY_MS = 20;
 const DEFAULT_CACHE_PERSIST_RETRY_MAX_DELAY_MS = 500;
+const DEFAULT_ISSUE_SOURCE_OF_TRUTH: IssueSourceOfTruth = 'last_write';
+
+type NormalizedIssueSource = 'github' | 'bit' | null;
 
 function nowEpochSec(): number {
   return Math.floor(Date.now() / 1000);
@@ -114,11 +119,58 @@ function describeError(error: unknown): string {
   return String(error);
 }
 
+function normalizeIssueSourceOfTruth(raw: IssueSourceOfTruth | undefined): IssueSourceOfTruth {
+  return raw === 'github' || raw === 'bit' ? raw : DEFAULT_ISSUE_SOURCE_OF_TRUTH;
+}
+
+function parseIssueSource(payload: JsonValue): NormalizedIssueSource {
+  if (!isObjectRecord(payload)) return null;
+  if (typeof payload.source !== 'string') return null;
+  const source = payload.source.trim().toLowerCase();
+  if (source === 'github') return 'github';
+  if (source === 'bit') return 'bit';
+  return null;
+}
+
+function isOlderByTimestamp(
+  incomingSourceUpdatedAtMs: number | null,
+  existingSourceUpdatedAtMs: number | null,
+): boolean {
+  return incomingSourceUpdatedAtMs !== null &&
+    existingSourceUpdatedAtMs !== null &&
+    incomingSourceUpdatedAtMs < existingSourceUpdatedAtMs;
+}
+
+function shouldUpdateSnapshotBySourcePolicy(args: {
+  policy: IssueSourceOfTruth;
+  incomingSource: NormalizedIssueSource;
+  existingSource: NormalizedIssueSource;
+  incomingSourceUpdatedAtMs: number | null;
+  existingSourceUpdatedAtMs: number | null;
+}): boolean {
+  if (isOlderByTimestamp(args.incomingSourceUpdatedAtMs, args.existingSourceUpdatedAtMs)) {
+    return false;
+  }
+  if (args.policy === 'last_write') {
+    return true;
+  }
+
+  const preferred = args.policy;
+  if (args.existingSource === preferred && args.incomingSource !== preferred) {
+    return false;
+  }
+  if (args.existingSource !== preferred && args.incomingSource === preferred) {
+    return true;
+  }
+  return true;
+}
+
 export function createRelayCacheAdapter(options: RelayCacheAdapterOptions): RelayCacheAdapter {
   const cacheStore = options.cacheStore;
   const isValidRoomName = options.isValidRoomName;
   const isValidTopic = options.isValidTopic;
   const nowSec = options.nowSec ?? nowEpochSec;
+  const issueSourceOfTruth = normalizeIssueSourceOfTruth(options.issueSourceOfTruth);
   const issueProjectionValidators: IssueProjectionValidators = {
     isValidRoomName,
     isValidTopic,
@@ -263,13 +315,16 @@ export function createRelayCacheAdapter(options: RelayCacheAdapterOptions): Rela
       );
       if (existing) {
         const parsed = parseIssueCacheSnapshotRecord(existing, issueProjectionValidators);
-        if (
-          parsed &&
-          parsed.source_updated_at_ms !== null &&
-          sourceUpdatedAtMs !== null &&
-          sourceUpdatedAtMs < parsed.source_updated_at_ms
-        ) {
-          shouldUpdateSnapshot = false;
+        if (parsed) {
+          const existingSource = parseIssueSource(parsed.envelope.payload);
+          const incomingSource = parseIssueSource(envelope.payload);
+          shouldUpdateSnapshot = shouldUpdateSnapshotBySourcePolicy({
+            policy: issueSourceOfTruth,
+            incomingSource,
+            existingSource,
+            incomingSourceUpdatedAtMs: sourceUpdatedAtMs,
+            existingSourceUpdatedAtMs: parsed.source_updated_at_ms,
+          });
         }
       }
     } catch {

src/runtime_config.ts

@@ -88,6 +88,14 @@ function parseOptionalString(raw: string | undefined): string | null {
   return value.length > 0 ? value : null;
 }
 
+function parseIssueSourceOfTruth(
+  raw: string | undefined,
+): 'last_write' | 'github' | 'bit' {
+  const value = (raw ?? '').trim().toLowerCase();
+  if (value === 'github' || value === 'bit') return value;
+  return 'last_write';
+}
+
 function parseRoomTokens(raw: string | undefined): Record<string, string> {
   if (typeof raw !== 'string' || raw.trim().length === 0) return {};
   try {
@@ -294,6 +302,8 @@ export function parseMemoryRelayOptionsFromEnv(getEnv: EnvGetter): MemoryRelayOp
 
   return {
     authToken: getEnv('BIT_RELAY_AUTH_TOKEN') ?? undefined,
+    peerAuthToken: getEnv('RELAY_PEER_AUTH_TOKEN') ?? undefined,
+    issueSourceOfTruth: parseIssueSourceOfTruth(getEnv('RELAY_ISSUE_SOURCE_OF_TRUTH')),
     maxMessagesPerRoom: parsePositiveInt(
       getEnv('RELAY_MAX_MESSAGES_PER_ROOM'),
       DEFAULT_MAX_MESSAGES_PER_ROOM,