From 3cb8b407542274c9a382f4e2b154771e7129154a Mon Sep 17 00:00:00 2001 From: "claude[bot]" <41898282+claude[bot]@users.noreply.github.com> Date: Sat, 11 Apr 2026 00:06:25 +0000 Subject: [PATCH] feat: release & developer experience (issue #16) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Packaging & Deployment: - Dockerfile for OpenThreads server (multi-stage Bun build) - docker-compose.yml production profile (app service, --profile production) - Helm chart scaffold (deploy/helm/openthreads/) with deployment, service, ingress, secret, HPA, serviceaccount templates - create-openthreads scaffold package (bunx create-openthreads ) Observability: - Structured JSON logger with configurable LOG_LEVEL + LOG_FORMAT env vars - Prometheus-compatible metrics endpoint (GET /api/metrics) with counters for messages in/out, A2H intents, fanout, HTTP requests, active threads - OpenTelemetry tracing setup in instrumentation.ts (enabled via OTEL_EXPORTER_OTLP_ENDPOINT, gracefully skipped if SDK not installed) A2H Layer 2 preparation: - Stub types for Layer 2 intents: POLICY, REVOKE, DELEGATE, SCOPE - isLayer1Intent() / isLayer2Intent() type guards - IntentHandlerRegistry extension point in reply-engine/intent-handler.ts for registering custom handlers (Layer 2 or overrides) - Export all new types from @openthreads/core Documentation: - docs/self-hosting.md (Docker, env vars, MongoDB, production checklist) - docs/adapter-authoring.md (ChannelAdapter interface guide) - docs/channels/slack.md, docs/channels/telegram.md Examples: - examples/plain-http/ — minimal Bun webhook consumer with cURL snippets - examples/n8n/ — n8n webhook node integration guide - examples/langgraph/ — LangGraph agent with A2H AUTHORIZE flow (Python) Co-authored-by: claude[bot] --- .env.example | 13 + Dockerfile | 53 ++++ deploy/helm/openthreads/Chart.yaml | 21 ++ .../helm/openthreads/templates/_helpers.tpl | 60 +++++ .../openthreads/templates/deployment.yaml | 74 ++++++ deploy/helm/openthreads/templates/hpa.yaml | 22 ++ .../helm/openthreads/templates/ingress.yaml | 35 +++ deploy/helm/openthreads/templates/secret.yaml | 15 ++ .../helm/openthreads/templates/service.yaml | 15 ++ .../openthreads/templates/serviceaccount.yaml | 13 + deploy/helm/openthreads/values.yaml | 136 ++++++++++ docker-compose.yml | 48 ++++ docs/adapter-authoring.md | 147 +++++++++++ docs/channels/slack.md | 77 ++++++ docs/channels/telegram.md | 62 +++++ docs/self-hosting.md | 169 ++++++++++++ examples/langgraph/README.md | 69 +++++ examples/langgraph/agent.py | 245 ++++++++++++++++++ examples/n8n/README.md | 117 +++++++++ examples/plain-http/README.md | 76 ++++++ examples/plain-http/server.ts | 120 +++++++++ package.json | 3 +- packages/core/src/index.ts | 26 +- .../core/src/reply-engine/intent-handler.ts | 133 ++++++++++ packages/core/src/types/a2h.ts | 68 ++++- packages/create-openthreads/package.json | 18 ++ packages/create-openthreads/src/index.ts | 180 +++++++++++++ packages/server/src/app/api/metrics/route.ts | 43 +++ packages/server/src/instrumentation.ts | 96 +++++-- packages/server/src/lib/logger.ts | 100 +++++-- packages/server/src/lib/metrics.ts | 141 ++++++++++ 31 files changed, 2354 insertions(+), 41 deletions(-) create mode 100644 Dockerfile create mode 100644 deploy/helm/openthreads/Chart.yaml create mode 100644 deploy/helm/openthreads/templates/_helpers.tpl create mode 100644 deploy/helm/openthreads/templates/deployment.yaml create mode 100644 deploy/helm/openthreads/templates/hpa.yaml create mode 100644 deploy/helm/openthreads/templates/ingress.yaml create mode 100644 deploy/helm/openthreads/templates/secret.yaml create mode 100644 deploy/helm/openthreads/templates/service.yaml create mode 100644 deploy/helm/openthreads/templates/serviceaccount.yaml create mode 100644 deploy/helm/openthreads/values.yaml create mode 100644 docs/adapter-authoring.md create mode 100644 docs/channels/slack.md create mode 100644 docs/channels/telegram.md create mode 100644 docs/self-hosting.md create mode 100644 examples/langgraph/README.md create mode 100644 examples/langgraph/agent.py create mode 100644 examples/n8n/README.md create mode 100644 examples/plain-http/README.md create mode 100644 examples/plain-http/server.ts create mode 100644 packages/core/src/reply-engine/intent-handler.ts create mode 100644 packages/create-openthreads/package.json create mode 100644 packages/create-openthreads/src/index.ts create mode 100644 packages/server/src/app/api/metrics/route.ts create mode 100644 packages/server/src/lib/metrics.ts diff --git a/.env.example b/.env.example index ac88509..c9e32a2 100644 --- a/.env.example +++ b/.env.example @@ -7,6 +7,10 @@ PORT=3000 NODE_ENV=development +# Logging (structured JSON logging) +# LOG_LEVEL=info # debug | info | warn | error (default: info) +# LOG_FORMAT=text # json | text (default: json in production, text in development) + # MongoDB # When using Docker Compose: mongodb://openthreads:openthreads@localhost:27017/openthreads MONGODB_URI=mongodb://openthreads:openthreads@localhost:27017/openthreads @@ -53,3 +57,12 @@ OPENTHREADS_BASE_URL=http://localhost:3000 # TRUST_JWS_ALGORITHM=RS256 # TRUST_PRIVATE_KEY_PATH=./keys/private.pem # TRUST_PUBLIC_KEY_PATH=./keys/public.pem + +# ============================================================================= +# Observability (optional) +# ============================================================================= + +# OpenTelemetry — set OTEL_EXPORTER_OTLP_ENDPOINT to enable tracing +# OTEL_EXPORTER_OTLP_ENDPOINT=http://otel-collector:4318 +# OTEL_SERVICE_NAME=openthreads +# OTEL_SDK_DISABLED=false diff --git a/Dockerfile b/Dockerfile new file mode 100644 index 0000000..e7d6822 --- /dev/null +++ b/Dockerfile @@ -0,0 +1,53 @@ +# ─── Build stage ───────────────────────────────────────────────────────────── +FROM oven/bun:1.2 AS builder + +WORKDIR /app + +# Copy workspace manifests first for layer caching +COPY package.json bun.lockb* ./ +COPY packages/core/package.json ./packages/core/ +COPY packages/server/package.json ./packages/server/ +COPY packages/storage/mongodb/package.json ./packages/storage/mongodb/ +COPY packages/trust/package.json ./packages/trust/ +COPY packages/channels/package.json ./packages/channels/ +COPY packages/channels/discord/package.json ./packages/channels/discord/ +COPY packages/channels/slack/package.json ./packages/channels/slack/ +COPY packages/channels/telegram/package.json ./packages/channels/telegram/ +COPY packages/channels/whatsapp/package.json ./packages/channels/whatsapp/ + +RUN bun install --frozen-lockfile + +# Copy source +COPY tsconfig.base.json tsconfig.json ./ +COPY packages ./packages + +# Build the Next.js server +WORKDIR /app/packages/server +ENV NEXT_TELEMETRY_DISABLED=1 +RUN bun run build + +# ─── Production stage ───────────────────────────────────────────────────────── +FROM oven/bun:1.2-slim AS runner + +WORKDIR /app + +ENV NODE_ENV=production +ENV NEXT_TELEMETRY_DISABLED=1 +ENV PORT=3000 + +RUN addgroup --system --gid 1001 nodejs && \ + adduser --system --uid 1001 nextjs + +# Copy built artifacts +COPY --from=builder /app/packages/server/.next/standalone ./ +COPY --from=builder /app/packages/server/.next/static ./packages/server/.next/static +COPY --from=builder /app/packages/server/public ./packages/server/public 2>/dev/null || true + +USER nextjs + +EXPOSE 3000 + +HEALTHCHECK --interval=30s --timeout=5s --start-period=15s --retries=3 \ + CMD bun -e "fetch('http://localhost:3000/api/health').then(r=>r.ok?process.exit(0):process.exit(1)).catch(()=>process.exit(1))" + +CMD ["bun", "packages/server/server.js"] diff --git a/deploy/helm/openthreads/Chart.yaml b/deploy/helm/openthreads/Chart.yaml new file mode 100644 index 0000000..2d97963 --- /dev/null +++ b/deploy/helm/openthreads/Chart.yaml @@ -0,0 +1,21 @@ +apiVersion: v2 +name: openthreads +description: > + OpenThreads — a unified messaging gateway with human-in-the-loop support. + Bridges Slack, Discord, Telegram, WhatsApp, and other channels to any + HTTP-based agent or service via the A2H protocol. +type: application +version: 0.1.0 +appVersion: "0.1.0" +keywords: + - openthreads + - a2h + - human-in-the-loop + - messaging + - webhook +home: https://github.com/deepducks/OpenThreads +sources: + - https://github.com/deepducks/OpenThreads +maintainers: + - name: DeepDucks + url: https://github.com/deepducks diff --git a/deploy/helm/openthreads/templates/_helpers.tpl b/deploy/helm/openthreads/templates/_helpers.tpl new file mode 100644 index 0000000..8d6003d --- /dev/null +++ b/deploy/helm/openthreads/templates/_helpers.tpl @@ -0,0 +1,60 @@ +{{/* +Expand the name of the chart. +*/}} +{{- define "openthreads.name" -}} +{{- default .Chart.Name .Values.nameOverride | trunc 63 | trimSuffix "-" }} +{{- end }} + +{{/* +Create a default fully qualified app name. +*/}} +{{- define "openthreads.fullname" -}} +{{- if .Values.fullnameOverride }} +{{- .Values.fullnameOverride | trunc 63 | trimSuffix "-" }} +{{- else }} +{{- $name := default .Chart.Name .Values.nameOverride }} +{{- if contains $name .Release.Name }} +{{- .Release.Name | trunc 63 | trimSuffix "-" }} +{{- else }} +{{- printf "%s-%s" .Release.Name $name | trunc 63 | trimSuffix "-" }} +{{- end }} +{{- end }} +{{- end }} + +{{/* +Create chart label. +*/}} +{{- define "openthreads.chart" -}} +{{- printf "%s-%s" .Chart.Name .Chart.Version | replace "+" "_" | trunc 63 | trimSuffix "-" }} +{{- end }} + +{{/* +Common labels. +*/}} +{{- define "openthreads.labels" -}} +helm.sh/chart: {{ include "openthreads.chart" . }} +{{ include "openthreads.selectorLabels" . }} +{{- if .Chart.AppVersion }} +app.kubernetes.io/version: {{ .Chart.AppVersion | quote }} +{{- end }} +app.kubernetes.io/managed-by: {{ .Release.Service }} +{{- end }} + +{{/* +Selector labels. +*/}} +{{- define "openthreads.selectorLabels" -}} +app.kubernetes.io/name: {{ include "openthreads.name" . }} +app.kubernetes.io/instance: {{ .Release.Name }} +{{- end }} + +{{/* +ServiceAccount name. +*/}} +{{- define "openthreads.serviceAccountName" -}} +{{- if .Values.serviceAccount.create }} +{{- default (include "openthreads.fullname" .) .Values.serviceAccount.name }} +{{- else }} +{{- default "default" .Values.serviceAccount.name }} +{{- end }} +{{- end }} diff --git a/deploy/helm/openthreads/templates/deployment.yaml b/deploy/helm/openthreads/templates/deployment.yaml new file mode 100644 index 0000000..bf9db38 --- /dev/null +++ b/deploy/helm/openthreads/templates/deployment.yaml @@ -0,0 +1,74 @@ +apiVersion: apps/v1 +kind: Deployment +metadata: + name: {{ include "openthreads.fullname" . }} + labels: + {{- include "openthreads.labels" . | nindent 4 }} +spec: + {{- if not .Values.autoscaling.enabled }} + replicas: {{ .Values.replicaCount }} + {{- end }} + selector: + matchLabels: + {{- include "openthreads.selectorLabels" . | nindent 6 }} + template: + metadata: + annotations: + checksum/secret: {{ include (print $.Template.BasePath "/secret.yaml") . | sha256sum }} + {{- if .Values.metrics.annotations }} + prometheus.io/scrape: "true" + prometheus.io/port: "{{ .Values.service.port }}" + prometheus.io/path: "/api/metrics" + {{- end }} + {{- with .Values.podAnnotations }} + {{- toYaml . | nindent 8 }} + {{- end }} + labels: + {{- include "openthreads.labels" . | nindent 8 }} + {{- with .Values.podLabels }} + {{- toYaml . | nindent 8 }} + {{- end }} + spec: + {{- with .Values.imagePullSecrets }} + imagePullSecrets: + {{- toYaml . | nindent 8 }} + {{- end }} + serviceAccountName: {{ include "openthreads.serviceAccountName" . }} + securityContext: + {{- toYaml .Values.podSecurityContext | nindent 8 }} + containers: + - name: {{ .Chart.Name }} + securityContext: + {{- toYaml .Values.securityContext | nindent 12 }} + image: "{{ .Values.image.repository }}:{{ .Values.image.tag | default .Chart.AppVersion }}" + imagePullPolicy: {{ .Values.image.pullPolicy }} + ports: + - name: http + containerPort: {{ .Values.service.port }} + protocol: TCP + envFrom: + - secretRef: + name: {{ if .Values.existingSecret }}{{ .Values.existingSecret }}{{ else }}{{ include "openthreads.fullname" . }}{{ end }} + env: + {{- range $key, $value := .Values.env }} + - name: {{ $key }} + value: {{ $value | quote }} + {{- end }} + livenessProbe: + {{- toYaml .Values.livenessProbe | nindent 12 }} + readinessProbe: + {{- toYaml .Values.readinessProbe | nindent 12 }} + resources: + {{- toYaml .Values.resources | nindent 12 }} + {{- with .Values.nodeSelector }} + nodeSelector: + {{- toYaml . | nindent 8 }} + {{- end }} + {{- with .Values.affinity }} + affinity: + {{- toYaml . | nindent 8 }} + {{- end }} + {{- with .Values.tolerations }} + tolerations: + {{- toYaml . | nindent 8 }} + {{- end }} diff --git a/deploy/helm/openthreads/templates/hpa.yaml b/deploy/helm/openthreads/templates/hpa.yaml new file mode 100644 index 0000000..6dfdd03 --- /dev/null +++ b/deploy/helm/openthreads/templates/hpa.yaml @@ -0,0 +1,22 @@ +{{- if .Values.autoscaling.enabled }} +apiVersion: autoscaling/v2 +kind: HorizontalPodAutoscaler +metadata: + name: {{ include "openthreads.fullname" . }} + labels: + {{- include "openthreads.labels" . | nindent 4 }} +spec: + scaleTargetRef: + apiVersion: apps/v1 + kind: Deployment + name: {{ include "openthreads.fullname" . }} + minReplicas: {{ .Values.autoscaling.minReplicas }} + maxReplicas: {{ .Values.autoscaling.maxReplicas }} + metrics: + - type: Resource + resource: + name: cpu + target: + type: Utilization + averageUtilization: {{ .Values.autoscaling.targetCPUUtilizationPercentage }} +{{- end }} diff --git a/deploy/helm/openthreads/templates/ingress.yaml b/deploy/helm/openthreads/templates/ingress.yaml new file mode 100644 index 0000000..ae51274 --- /dev/null +++ b/deploy/helm/openthreads/templates/ingress.yaml @@ -0,0 +1,35 @@ +{{- if .Values.ingress.enabled -}} +apiVersion: networking.k8s.io/v1 +kind: Ingress +metadata: + name: {{ include "openthreads.fullname" . }} + labels: + {{- include "openthreads.labels" . | nindent 4 }} + {{- with .Values.ingress.annotations }} + annotations: + {{- toYaml . | nindent 4 }} + {{- end }} +spec: + {{- if .Values.ingress.className }} + ingressClassName: {{ .Values.ingress.className }} + {{- end }} + {{- if .Values.ingress.tls }} + tls: + {{- toYaml .Values.ingress.tls | nindent 4 }} + {{- end }} + rules: + {{- range .Values.ingress.hosts }} + - host: {{ .host | quote }} + http: + paths: + {{- range .paths }} + - path: {{ .path }} + pathType: {{ .pathType }} + backend: + service: + name: {{ include "openthreads.fullname" $ }} + port: + number: {{ $.Values.service.port }} + {{- end }} + {{- end }} +{{- end }} diff --git a/deploy/helm/openthreads/templates/secret.yaml b/deploy/helm/openthreads/templates/secret.yaml new file mode 100644 index 0000000..6db4451 --- /dev/null +++ b/deploy/helm/openthreads/templates/secret.yaml @@ -0,0 +1,15 @@ +{{- if not .Values.existingSecret }} +apiVersion: v1 +kind: Secret +metadata: + name: {{ include "openthreads.fullname" . }} + labels: + {{- include "openthreads.labels" . | nindent 4 }} +type: Opaque +stringData: + {{- range $key, $value := .Values.secrets }} + {{- if $value }} + {{ $key }}: {{ $value | quote }} + {{- end }} + {{- end }} +{{- end }} diff --git a/deploy/helm/openthreads/templates/service.yaml b/deploy/helm/openthreads/templates/service.yaml new file mode 100644 index 0000000..0c3d9ad --- /dev/null +++ b/deploy/helm/openthreads/templates/service.yaml @@ -0,0 +1,15 @@ +apiVersion: v1 +kind: Service +metadata: + name: {{ include "openthreads.fullname" . }} + labels: + {{- include "openthreads.labels" . | nindent 4 }} +spec: + type: {{ .Values.service.type }} + ports: + - port: {{ .Values.service.port }} + targetPort: http + protocol: TCP + name: http + selector: + {{- include "openthreads.selectorLabels" . | nindent 4 }} diff --git a/deploy/helm/openthreads/templates/serviceaccount.yaml b/deploy/helm/openthreads/templates/serviceaccount.yaml new file mode 100644 index 0000000..d18712e --- /dev/null +++ b/deploy/helm/openthreads/templates/serviceaccount.yaml @@ -0,0 +1,13 @@ +{{- if .Values.serviceAccount.create -}} +apiVersion: v1 +kind: ServiceAccount +metadata: + name: {{ include "openthreads.serviceAccountName" . }} + labels: + {{- include "openthreads.labels" . | nindent 4 }} + {{- with .Values.serviceAccount.annotations }} + annotations: + {{- toYaml . | nindent 4 }} + {{- end }} +automountServiceAccountToken: {{ .Values.serviceAccount.automount }} +{{- end }} diff --git a/deploy/helm/openthreads/values.yaml b/deploy/helm/openthreads/values.yaml new file mode 100644 index 0000000..3578ab3 --- /dev/null +++ b/deploy/helm/openthreads/values.yaml @@ -0,0 +1,136 @@ +# Default values for the OpenThreads Helm chart. +# Override these in a custom values file: helm install openthreads ./openthreads -f my-values.yaml + +# ─── Image ──────────────────────────────────────────────────────────────────── +image: + repository: ghcr.io/deepducks/openthreads + tag: "" # Defaults to .Chart.AppVersion when empty + pullPolicy: IfNotPresent + +imagePullSecrets: [] +nameOverride: "" +fullnameOverride: "" + +# ─── Deployment ─────────────────────────────────────────────────────────────── +replicaCount: 1 + +podAnnotations: {} +podLabels: {} + +podSecurityContext: + runAsNonRoot: true + runAsUser: 1001 + +securityContext: + allowPrivilegeEscalation: false + capabilities: + drop: [ALL] + readOnlyRootFilesystem: true + +resources: + requests: + cpu: 100m + memory: 256Mi + limits: + cpu: 500m + memory: 512Mi + +nodeSelector: {} +tolerations: [] +affinity: {} + +# ─── Service ────────────────────────────────────────────────────────────────── +service: + type: ClusterIP + port: 3000 + +# ─── Ingress ────────────────────────────────────────────────────────────────── +ingress: + enabled: false + className: "" + annotations: {} + hosts: + - host: openthreads.example.com + paths: + - path: / + pathType: Prefix + tls: [] + +# ─── Environment ────────────────────────────────────────────────────────────── +env: + NODE_ENV: production + PORT: "3000" + LOG_LEVEL: info + LOG_FORMAT: json + REPLY_TOKEN_TTL: "86400" + +# ─── Secrets (stored in a Kubernetes Secret — do NOT put plain values here) ─── +# Reference an existing secret instead: +# existingSecret: my-openthreads-secret +# Or let the chart create one from the values below (not recommended for prod): +secrets: + # Required + JWT_SECRET: "" + MONGODB_URI: "" + # Optional + MANAGEMENT_API_KEY: "" + OPENTHREADS_BASE_URL: "" + # Channel credentials + SLACK_BOT_TOKEN: "" + SLACK_SIGNING_SECRET: "" + SLACK_APP_TOKEN: "" + TELEGRAM_BOT_TOKEN: "" + DISCORD_BOT_TOKEN: "" + DISCORD_CLIENT_ID: "" + # Trust layer + TRUST_LAYER_ENABLED: "false" + +# Reference an existing secret instead of creating one from 'secrets' above. +existingSecret: "" + +# ─── Health check ───────────────────────────────────────────────────────────── +livenessProbe: + httpGet: + path: /api/health + port: 3000 + initialDelaySeconds: 15 + periodSeconds: 30 + timeoutSeconds: 5 + failureThreshold: 3 + +readinessProbe: + httpGet: + path: /api/health + port: 3000 + initialDelaySeconds: 5 + periodSeconds: 10 + timeoutSeconds: 3 + failureThreshold: 3 + +# ─── MongoDB (sub-chart, disabled by default — use an external MongoDB in prod) ─ +mongodb: + enabled: false + auth: + rootPassword: "" + username: openthreads + password: openthreads + database: openthreads + +# ─── ServiceAccount ─────────────────────────────────────────────────────────── +serviceAccount: + create: true + automount: true + annotations: {} + name: "" + +# ─── Autoscaling ────────────────────────────────────────────────────────────── +autoscaling: + enabled: false + minReplicas: 1 + maxReplicas: 5 + targetCPUUtilizationPercentage: 70 + +# ─── Prometheus scraping ────────────────────────────────────────────────────── +metrics: + # Add Prometheus scrape annotations to the pod + annotations: true diff --git a/docker-compose.yml b/docker-compose.yml index 789489e..b0b1e16 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -1,5 +1,14 @@ version: '3.8' +# ─── Profiles ───────────────────────────────────────────────────────────────── +# +# (default) mongodb only — use for local dev with `bun run dev` +# production full stack: mongodb + openthreads app +# +# Usage: +# docker compose up # dev: MongoDB only +# docker compose --profile production up -d # prod: full stack + services: mongodb: image: mongo:7.0 @@ -20,6 +29,45 @@ services: start_period: 10s restart: unless-stopped + app: + profiles: [production] + image: ghcr.io/deepducks/openthreads:latest + build: + context: . + dockerfile: Dockerfile + container_name: openthreads-app + ports: + - '${PORT:-3000}:3000' + environment: + NODE_ENV: production + PORT: 3000 + MONGODB_URI: mongodb://openthreads:openthreads@mongodb:27017/openthreads + JWT_SECRET: ${JWT_SECRET:?JWT_SECRET must be set} + MANAGEMENT_API_KEY: ${MANAGEMENT_API_KEY:-} + OPENTHREADS_BASE_URL: ${OPENTHREADS_BASE_URL:-http://localhost:3000} + REPLY_TOKEN_TTL: ${REPLY_TOKEN_TTL:-86400} + LOG_LEVEL: ${LOG_LEVEL:-info} + LOG_FORMAT: ${LOG_FORMAT:-json} + # Channel credentials (set the ones you need) + SLACK_BOT_TOKEN: ${SLACK_BOT_TOKEN:-} + SLACK_SIGNING_SECRET: ${SLACK_SIGNING_SECRET:-} + SLACK_APP_TOKEN: ${SLACK_APP_TOKEN:-} + TELEGRAM_BOT_TOKEN: ${TELEGRAM_BOT_TOKEN:-} + DISCORD_BOT_TOKEN: ${DISCORD_BOT_TOKEN:-} + DISCORD_CLIENT_ID: ${DISCORD_CLIENT_ID:-} + # Trust layer (optional) + TRUST_LAYER_ENABLED: ${TRUST_LAYER_ENABLED:-false} + depends_on: + mongodb: + condition: service_healthy + restart: unless-stopped + healthcheck: + test: ['CMD-SHELL', 'curl -f http://localhost:3000/api/health || exit 1'] + interval: 30s + timeout: 5s + retries: 3 + start_period: 15s + volumes: mongodb_data: driver: local diff --git a/docs/adapter-authoring.md b/docs/adapter-authoring.md new file mode 100644 index 0000000..c2eb6da --- /dev/null +++ b/docs/adapter-authoring.md @@ -0,0 +1,147 @@ +# Custom Channel Adapter Guide + +OpenThreads uses a `ChannelAdapter` interface to abstract platform-specific +messaging. This guide explains how to implement a custom adapter for any +platform not natively supported. + +## When do you need a custom adapter? + +- Your platform isn't supported out of the box (WhatsApp via Baileys, Signal, Matrix, etc.) +- You need custom rendering for A2H intents on a supported platform +- You're integrating with an internal messaging system + +## The ChannelAdapter interface + +```typescript +import type { ChannelAdapter } from '@openthreads/core'; + +export class MyAdapter implements ChannelAdapter { + // Report what your platform supports + capabilities(): ChannelCapabilities { ... } + + // Set up webhooks / subscriptions when a channel is registered + async register(config: ChannelConfig): Promise { ... } + + // Send a message to a target (channel, group, user) + async sendMessage(target: string, message: ...): Promise { ... } + + // Render a Chat SDK message in platform-native format + async renderChatSDK(message: ChatSDKMessage, capabilities): Promise { ... } + + // Render an A2H intent as inline interactive elements (buttons, menus) + // Only called for Method 1 (inline rendering) + async renderA2HInline(intent: A2HMessage, capabilities): Promise { ... } + + // Capture a free-text response from the human (Method 2) + async captureResponse(thread: Thread, turn: Turn): Promise { ... } +} +``` + +## Step-by-step: implement a minimal adapter + +### 1. Create a new package + +``` +packages/channels/my-platform/ + src/ + adapter.ts # ChannelAdapter implementation + index.ts # public exports + package.json +``` + +### 2. Declare capabilities + +```typescript +capabilities(): ChannelCapabilities { + return { + threads: false, // does your platform have native threads? + buttons: true, // can you render interactive buttons? + selectMenus: false, // can you render dropdown menus? + replyMessages: true, // can senders reply to specific messages? + dms: true, // does it support DMs? + fileUpload: false, // can you upload files? + }; +} +``` + +The Reply Engine uses these flags to select the best method (1-4) for each +A2H intent. Reporting wrong capabilities leads to degraded UX. + +### 3. Implement `renderChatSDK` + +Map the Chat SDK message to your platform's native format: + +```typescript +async renderChatSDK( + message: ChatSDKMessage, + _capabilities: ChannelCapabilities, +): Promise { + return { + text: message.text ?? '', + // Add platform-specific fields + }; +} +``` + +### 4. Implement `renderA2HInline` (Method 1) + +For `AUTHORIZE` (approve/deny) and `COLLECT` with closed options: + +```typescript +async renderA2HInline( + intent: A2HMessage, + _capabilities: ChannelCapabilities, +): Promise { + if (intent.intent === 'AUTHORIZE') { + return { + text: intent.description ?? 'Action requires your approval', + // Platform-specific button payload + buttons: [ + { id: 'approve', text: 'Approve', value: 'true' }, + { id: 'deny', text: 'Deny', value: 'false' }, + ], + }; + } + // Handle COLLECT with options... +} +``` + +### 5. Implement `captureResponse` (Method 2) + +For free-text collection via thread/reply: + +```typescript +async captureResponse(thread: Thread, turn: Turn): Promise { + // Set up a listener for the next message in this thread from the sender + // This depends heavily on your platform's event system + return new Promise((resolve) => { + // ... platform-specific listener + }); +} +``` + +### 6. Implement `register` + +```typescript +async register(config: ChannelConfig): Promise { + // Store config + // Register webhooks with the platform + // Subscribe to events +} +``` + +## Example: WhatsApp via Baileys + +See `packages/channels/whatsapp/` for a real-world example using the +[Baileys](https://github.com/WhiskeySockets/Baileys) library. + +## Publishing your adapter + +Adapters are standalone npm packages. Publish yours and users can install it +alongside OpenThreads: + +```bash +npm install @my-org/openthreads-adapter-myplatform +``` + +Then register it in the OpenThreads channel configuration. diff --git a/docs/channels/slack.md b/docs/channels/slack.md new file mode 100644 index 0000000..6251c57 --- /dev/null +++ b/docs/channels/slack.md @@ -0,0 +1,77 @@ +# Channel Setup: Slack + +## Prerequisites + +- A Slack workspace where you have admin permissions +- OpenThreads running at a publicly accessible HTTPS URL (required for webhooks) + +## Step 1 — Create a Slack App + +1. Go to [api.slack.com/apps](https://api.slack.com/apps) → **Create New App** → **From scratch** +2. Name your app (e.g., `OpenThreads`) and choose your workspace +3. Note your **App ID** and **Signing Secret** (under **Basic Information**) + +## Step 2 — Configure OAuth scopes + +Under **OAuth & Permissions** → **Scopes** → **Bot Token Scopes**, add: + +| Scope | Purpose | +|---|---| +| `channels:history` | Read messages in public channels | +| `channels:read` | List channels | +| `chat:write` | Post messages | +| `im:history` | Read DMs | +| `im:write` | Send DMs | +| `groups:history` | Read private channels | +| `users:read` | Look up user info | +| `app_mentions:read` | Receive mention events | +| `reactions:write` | Add emoji reactions | + +For interactive components (A2H Method 1 — buttons): + +| Scope | Purpose | +|---|---| +| `chat:write.customize` | Post as custom names/icons | + +## Step 3 — Enable Event Subscriptions + +Under **Event Subscriptions**: +1. Toggle **Enable Events** on +2. Set **Request URL** to: `https://your-openthreads.example.com/webhook/` +3. Subscribe to bot events: + - `message.channels` + - `message.im` + - `message.groups` + - `app_mention` + +## Step 4 — Enable Interactivity (for A2H Method 1) + +Under **Interactivity & Shortcuts**: +1. Toggle **Interactivity** on +2. Set **Request URL** to: `https://your-openthreads.example.com/webhook//interactive` + +## Step 5 — Install the app + +Under **OAuth & Permissions** → **Install to Workspace**. Copy the **Bot User OAuth Token** (starts with `xoxb-`). + +## Step 6 — Register in OpenThreads + +```bash +curl -s -X POST http://localhost:3000/api/channels \ + -H 'Content-Type: application/json' \ + -H "Authorization: Bearer $MANAGEMENT_API_KEY" \ + -d '{ + "id": "my-slack", + "name": "My Slack", + "platform": "slack", + "credentialsRef": "slack-main" + }' | jq . +``` + +Set environment variables: + +```bash +SLACK_BOT_TOKEN=xoxb-your-bot-token +SLACK_SIGNING_SECRET=your-signing-secret +SLACK_APP_TOKEN=xapp-your-app-token # for Socket Mode +``` diff --git a/docs/channels/telegram.md b/docs/channels/telegram.md new file mode 100644 index 0000000..e694541 --- /dev/null +++ b/docs/channels/telegram.md @@ -0,0 +1,62 @@ +# Channel Setup: Telegram + +## Prerequisites + +- A Telegram account +- OpenThreads running at a publicly accessible HTTPS URL (required for webhooks) + +## Step 1 — Create a bot + +1. Open Telegram and message [@BotFather](https://t.me/botfather) +2. Send `/newbot` and follow the prompts +3. Copy the **bot token** (format: `123456789:ABCdefGhIJKlmNoPQRstUVwxyZ`) + +## Step 2 — Configure the bot (optional) + +``` +/setdescription — add a description +/setuserpic — add a profile photo +/setcommands — define bot commands (e.g., /help) +``` + +## Step 3 — Register webhook + +OpenThreads registers the webhook automatically when you start the server with +`TELEGRAM_BOT_TOKEN` set. You can verify: + +```bash +curl https://api.telegram.org/bot/getWebhookInfo +``` + +## Step 4 — Register in OpenThreads + +```bash +curl -s -X POST http://localhost:3000/api/channels \ + -H 'Content-Type: application/json' \ + -H "Authorization: Bearer $MANAGEMENT_API_KEY" \ + -d '{ + "id": "my-telegram", + "name": "My Telegram Bot", + "platform": "telegram", + "credentialsRef": "telegram-main" + }' | jq . +``` + +Set environment variables: + +```bash +TELEGRAM_BOT_TOKEN=123456789:ABCdefGhIJKlmNoPQRstUVwxyZ +# Optional: restrict webhook updates to a secret token +TELEGRAM_WEBHOOK_SECRET=your-random-secret +``` + +## Telegram capabilities + +| Feature | Supported | +|---|---| +| Inline keyboards (buttons) | Yes — A2H Method 1 | +| Reply to message | Yes — A2H Method 2 | +| Native threads | No (groups only, not DMs) | +| File uploads | Yes | +| Group chats | Yes | +| DMs | Yes | diff --git a/docs/self-hosting.md b/docs/self-hosting.md new file mode 100644 index 0000000..a91c7ac --- /dev/null +++ b/docs/self-hosting.md @@ -0,0 +1,169 @@ +# Self-Hosting Guide + +OpenThreads is designed to be self-hosted. This guide covers Docker, environment +variable configuration, and MongoDB setup. + +## Prerequisites + +- Docker + Docker Compose (v2.x) +- A domain name with HTTPS (required for Slack/Discord webhooks in production) +- MongoDB 7.x (managed by Docker Compose or an external service) + +--- + +## Quick start with Docker Compose + +### 1. Clone or scaffold + +```bash +# Scaffold a new deployment: +bunx create-openthreads my-deployment +cd my-deployment + +# Or clone the repository: +git clone https://github.com/deepducks/OpenThreads.git +cd OpenThreads +``` + +### 2. Configure environment + +```bash +cp .env.example .env +# Edit .env — at minimum, set JWT_SECRET and MONGODB_URI +``` + +### 3. Start + +```bash +# Start MongoDB + OpenThreads +docker compose --profile production up -d + +# Check logs +docker compose logs -f app + +# Health check +curl http://localhost:3000/api/health +``` + +--- + +## Environment Variables + +### Required + +| Variable | Description | Example | +|---|---|---| +| `MONGODB_URI` | MongoDB connection URI | `mongodb://user:pass@host:27017/openthreads` | +| `JWT_SECRET` | Secret for signing JWTs — use a long random string | `openssl rand -hex 32` | + +### Recommended for production + +| Variable | Description | Default | +|---|---|---| +| `MANAGEMENT_API_KEY` | Protects `/api/*` management endpoints | unset (open) | +| `OPENTHREADS_BASE_URL` | Public base URL (used in `replyTo` URLs) | `http://localhost:3000` | +| `NODE_ENV` | Set to `production` in prod | `development` | +| `LOG_LEVEL` | `debug` \| `info` \| `warn` \| `error` | `info` | +| `LOG_FORMAT` | `json` for structured logging, `text` for human-readable | `json` in prod | +| `REPLY_TOKEN_TTL` | `replyTo` token TTL in seconds | `86400` (24h) | + +### Channel credentials + +Set the credentials for each channel you want to enable. See the [channel setup guides](./channels/) for how to obtain these values. + +| Variable | Platform | +|---|---| +| `SLACK_BOT_TOKEN`, `SLACK_SIGNING_SECRET`, `SLACK_APP_TOKEN` | Slack | +| `TELEGRAM_BOT_TOKEN` | Telegram | +| `DISCORD_BOT_TOKEN`, `DISCORD_CLIENT_ID` | Discord | + +### Trust layer (optional) + +| Variable | Description | Default | +|---|---|---| +| `TRUST_LAYER_ENABLED` | Enable JWS signing and WebAuthn | `false` | +| `TRUST_JWS_ALGORITHM` | JWS algorithm | `RS256` | +| `TRUST_PRIVATE_KEY_PATH` | Path to private key PEM | — | +| `TRUST_PUBLIC_KEY_PATH` | Path to public key PEM | — | + +### OpenTelemetry (optional) + +| Variable | Description | +|---|---| +| `OTEL_EXPORTER_OTLP_ENDPOINT` | OTLP endpoint, e.g. `http://otel-collector:4318` | +| `OTEL_SERVICE_NAME` | Service name tag (default: `openthreads`) | +| `OTEL_SDK_DISABLED` | Set to `true` to disable tracing | + +--- + +## MongoDB Setup + +### Using the bundled Docker Compose MongoDB + +The included `docker-compose.yml` starts a MongoDB 7 instance with: +- Username: `openthreads` +- Password: `openthreads` +- Database: `openthreads` +- Data persisted in the `mongodb_data` Docker volume + +Connection string: `mongodb://openthreads:openthreads@localhost:27017/openthreads` + +### Using an external MongoDB + +Set `MONGODB_URI` to your connection string. OpenThreads creates indexes +automatically on first start. + +Recommended: MongoDB Atlas free tier for small deployments. + +### Indexes + +OpenThreads automatically ensures the following indexes on startup: +- `channels.id` (unique) +- `recipients.id` (unique) +- `threads.threadId` (unique), `threads.channelId+nativeThreadId` +- `turns.turnId` (unique), `turns.threadId+timestamp` +- `routes.id` (unique), `routes.priority` +- `tokens.value` (unique, with TTL) +- `audit_log.*` (several indexes) + +--- + +## Production checklist + +- [ ] `JWT_SECRET` is a long random string (not `change-me`) +- [ ] `MANAGEMENT_API_KEY` is set (protects admin API) +- [ ] `OPENTHREADS_BASE_URL` is your public HTTPS URL +- [ ] `NODE_ENV=production` +- [ ] `LOG_FORMAT=json` (for log aggregators like Loki/CloudWatch) +- [ ] TLS termination via reverse proxy (nginx, Caddy, Cloudflare Tunnel) +- [ ] MongoDB has authentication enabled and is not exposed publicly +- [ ] Prometheus scraping configured (if using `LOG_FORMAT=json`) + +--- + +## Reverse proxy setup + +### Caddy (recommended) + +```caddyfile +openthreads.example.com { + reverse_proxy localhost:3000 +} +``` + +### Nginx + +```nginx +server { + listen 443 ssl; + server_name openthreads.example.com; + + location / { + proxy_pass http://localhost:3000; + proxy_set_header Host $host; + proxy_set_header X-Real-IP $remote_addr; + proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for; + proxy_set_header X-Forwarded-Proto $scheme; + } +} +``` diff --git a/examples/langgraph/README.md b/examples/langgraph/README.md new file mode 100644 index 0000000..f29a55c --- /dev/null +++ b/examples/langgraph/README.md @@ -0,0 +1,69 @@ +# Example: LangGraph Integration + +Use OpenThreads as the human-in-the-loop channel for a LangGraph agent. When +the agent needs human input (approval, data collection, escalation), it sends +an A2H intent to OpenThreads via the `replyTo` URL and blocks until the human +responds. + +## Architecture + +``` +Human (Slack/Telegram/Discord) + │ inbound message + ▼ +OpenThreads ──envelope──► LangGraph agent + │ (processes, may interrupt) + │ POST replyTo A2H intent + ▼ +OpenThreads ──renders──► Human (approve/deny buttons, form, etc.) + │ human responds + ▼ +OpenThreads ──response──► LangGraph agent (interrupt resolves) +``` + +## Prerequisites + +```bash +pip install langgraph langchain-openai httpx +``` + +## Example agent + +See `agent.py` for a complete example of a LangGraph agent that: +1. Receives a task from OpenThreads +2. Runs an autonomous sub-task +3. Sends an `AUTHORIZE` intent to OpenThreads when it needs human approval +4. Resumes after the human responds + +## Key pattern + +```python +import httpx + +async def ask_human(reply_to: str, intent: dict) -> dict: + """ + Send an A2H intent to OpenThreads and wait for the human's response. + OpenThreads blocks the POST until the human responds (or the token expires). + """ + async with httpx.AsyncClient(timeout=300) as client: + response = await client.post( + reply_to, + json={"message": [intent]}, + ) + response.raise_for_status() + return response.json() + +# In your LangGraph node: +result = await ask_human( + reply_to=state["replyTo"], + intent={ + "intent": "AUTHORIZE", + "context": { + "action": "send-email", + "details": f"Send report to {recipient} (150KB attachment)" + } + } +) + +approved = result.get("responses", [{}])[0].get("response", False) +``` diff --git a/examples/langgraph/agent.py b/examples/langgraph/agent.py new file mode 100644 index 0000000..7904422 --- /dev/null +++ b/examples/langgraph/agent.py @@ -0,0 +1,245 @@ +""" +LangGraph + OpenThreads Integration Example. + +This example shows a LangGraph agent that: + 1. Receives a task envelope from OpenThreads (via its own HTTP server) + 2. Processes the task autonomously + 3. Sends an A2H AUTHORIZE intent to the human via OpenThreads + 4. Waits for the human's approval before completing the task + 5. Replies with the result + +Requirements: + pip install langgraph langchain-openai httpx fastapi uvicorn + +Run: + python agent.py +""" + +import asyncio +import json +import os +from typing import TypedDict, Annotated + +import httpx +from fastapi import FastAPI, Request +from fastapi.responses import JSONResponse + +try: + from langgraph.graph import StateGraph, END + from langgraph.graph.message import add_messages + from langchain_core.messages import HumanMessage, AIMessage + LANGGRAPH_AVAILABLE = True +except ImportError: + LANGGRAPH_AVAILABLE = False + print("LangGraph not installed. Install with: pip install langgraph langchain-openai") + +# ─── Agent state ────────────────────────────────────────────────────────────── + +class AgentState(TypedDict): + # OpenThreads envelope fields + thread_id: str + turn_id: str + reply_to: str + sender_name: str + # Task state + task: str + plan: str + approved: bool + result: str + + +# ─── A2H helper ─────────────────────────────────────────────────────────────── + +async def send_a2h_intent(reply_to: str, intent: dict, timeout: float = 300) -> dict: + """ + POST an A2H intent to OpenThreads' replyTo URL and return the response. + + OpenThreads renders the intent to the human (buttons, form, etc.) and + blocks the HTTP request until the human responds. The response contains + the human's answer (approved/denied, collected data, etc.). + + Args: + reply_to: The replyTo URL from the OpenThreads envelope. + intent: An A2H message dict (must contain 'intent' key). + timeout: Seconds to wait for the human response (default: 5 min). + + Returns: + The JSON response from OpenThreads, e.g.: + {"responses": [{"intent": "AUTHORIZE", "response": true, "respondedAt": "..."}]} + """ + async with httpx.AsyncClient(timeout=timeout) as client: + response = await client.post( + reply_to, + json={"message": [intent]}, + headers={"Content-Type": "application/json"}, + ) + response.raise_for_status() + return response.json() + + +async def send_text(reply_to: str, text: str) -> None: + """Send a simple text message via OpenThreads.""" + async with httpx.AsyncClient(timeout=30) as client: + await client.post( + reply_to, + json={"message": {"text": text}}, + headers={"Content-Type": "application/json"}, + ) + + +# ─── LangGraph agent ────────────────────────────────────────────────────────── + +async def plan_task(state: AgentState) -> dict: + """Generate a plan for the task (runs autonomously).""" + task = state["task"] + # In a real agent, call an LLM here + plan = f"Plan for '{task}': Step 1 → analyse, Step 2 → execute, Step 3 → report" + print(f"[agent] planned: {plan}") + return {"plan": plan} + + +async def request_approval(state: AgentState) -> dict: + """Send an AUTHORIZE intent to the human and wait for approval.""" + print(f"[agent] requesting approval via OpenThreads...") + + try: + result = await send_a2h_intent( + reply_to=state["reply_to"], + intent={ + "intent": "AUTHORIZE", + "context": { + "action": "execute-plan", + "details": state["plan"], + "requestedBy": "LangGraph agent", + }, + "description": "Please review and approve the plan before execution.", + }, + ) + + responses = result.get("responses", []) + approved = bool(responses[0].get("response", False)) if responses else False + print(f"[agent] human {'approved' if approved else 'denied'}") + return {"approved": approved} + + except httpx.TimeoutException: + print("[agent] approval request timed out") + return {"approved": False} + + +async def execute_task(state: AgentState) -> dict: + """Execute the task (only runs if approved).""" + if not state["approved"]: + return {"result": "Task was not approved by the human."} + + # Simulate task execution + await asyncio.sleep(1) + result = f"Task completed successfully. Plan executed: {state['plan']}" + print(f"[agent] {result}") + return {"result": result} + + +async def send_result(state: AgentState) -> dict: + """Send the final result back to the human via OpenThreads.""" + await send_text(state["reply_to"], state["result"]) + print("[agent] result sent to human") + return {} + + +def should_execute(state: AgentState) -> str: + return "execute" if state.get("approved") else "send_result" + + +def build_graph(): + """Build and compile the LangGraph state machine.""" + graph = StateGraph(AgentState) + + graph.add_node("plan", plan_task) + graph.add_node("request_approval", request_approval) + graph.add_node("execute", execute_task) + graph.add_node("send_result", send_result) + + graph.set_entry_point("plan") + graph.add_edge("plan", "request_approval") + graph.add_conditional_edges( + "request_approval", + should_execute, + {"execute": "execute", "send_result": "send_result"}, + ) + graph.add_edge("execute", "send_result") + graph.add_edge("send_result", END) + + return graph.compile() + + +# ─── HTTP server (receives OpenThreads envelopes) ──────────────────────────── + +app = FastAPI(title="LangGraph + OpenThreads Agent") + +@app.post("/inbound") +async def inbound(request: Request): + """Receive an OpenThreads envelope and kick off the LangGraph agent.""" + envelope = await request.json() + + thread_id = envelope.get("threadId", "") + turn_id = envelope.get("turnId", "") + reply_to = envelope.get("replyTo", "") + source = envelope.get("source", {}) + sender_name = source.get("sender", {}).get("name", "unknown") + + # Extract the task text from the message + message = envelope.get("message", []) + if isinstance(message, dict): + message = [message] + task = " ".join( + m.get("text", "") for m in message if isinstance(m, dict) and "text" in m + ).strip() or "Do something useful" + + print(f"[inbound] task='{task}' from={sender_name} thread={thread_id}") + + # Acknowledge immediately + asyncio.create_task(run_agent(thread_id, turn_id, reply_to, sender_name, task)) + return JSONResponse({"ok": True}) + + +@app.get("/health") +async def health(): + return {"status": "ok"} + + +async def run_agent( + thread_id: str, + turn_id: str, + reply_to: str, + sender_name: str, + task: str, +) -> None: + if not LANGGRAPH_AVAILABLE: + print("[agent] LangGraph not available — sending error reply") + await send_text(reply_to, "Error: LangGraph is not installed.") + return + + graph = build_graph() + initial_state: AgentState = { + "thread_id": thread_id, + "turn_id": turn_id, + "reply_to": reply_to, + "sender_name": sender_name, + "task": task, + "plan": "", + "approved": False, + "result": "", + } + try: + await graph.ainvoke(initial_state) + except Exception as exc: + print(f"[agent] error: {exc}") + await send_text(reply_to, f"Error processing task: {exc}") + + +# ─── Entry point ───────────────────────────────────────────────────────────── + +if __name__ == "__main__": + import uvicorn + port = int(os.environ.get("PORT", 4001)) + print(f"[server] LangGraph agent listening on http://localhost:{port}") + uvicorn.run(app, host="0.0.0.0", port=port) diff --git a/examples/n8n/README.md b/examples/n8n/README.md new file mode 100644 index 0000000..3fc83de --- /dev/null +++ b/examples/n8n/README.md @@ -0,0 +1,117 @@ +# Example: n8n Integration + +Receive OpenThreads message envelopes in an n8n workflow and reply via the +`replyTo` URL — no custom code required. + +## Architecture + +``` +Human (Slack) → OpenThreads → n8n Webhook node + ↓ + [your n8n workflow] + ↓ +Human (Slack) ← OpenThreads ← HTTP Request node (POST replyTo) +``` + +## Step 1 — Create an n8n Webhook + +1. In your n8n workflow, add a **Webhook** node. +2. Set **HTTP Method** to `POST`. +3. Set **Response Mode** to `Immediately` (return 200 at once; process async). +4. Copy the **Webhook URL** (e.g., `https://your-n8n.example.com/webhook/openthreads`). + +## Step 2 — Create an OpenThreads Route + +Register a recipient in OpenThreads that points to your n8n webhook URL: + +```bash +curl -s -X POST http://localhost:3000/api/recipients \ + -H 'Content-Type: application/json' \ + -H "Authorization: Bearer $MANAGEMENT_API_KEY" \ + -d '{ + "id": "n8n-workflow", + "name": "n8n Workflow", + "webhookUrl": "https://your-n8n.example.com/webhook/openthreads" + }' | jq . +``` + +Then create a route to forward messages to it: + +```bash +curl -s -X POST http://localhost:3000/api/routes \ + -H 'Content-Type: application/json' \ + -H "Authorization: Bearer $MANAGEMENT_API_KEY" \ + -d '{ + "id": "slack-to-n8n", + "name": "Slack → n8n", + "recipientId": "n8n-workflow", + "criteria": { "channelId": "my-slack-channel" }, + "enabled": true, + "priority": 1 + }' | jq . +``` + +## Step 3 — Build your n8n workflow + +The Webhook node receives the OpenThreads envelope: + +```json +{ + "threadId": "ot_thr_abc123", + "turnId": "ot_turn_001", + "replyTo": "http://localhost:3000/send/channel/my-slack/target/C0123/thread/ot_thr_abc123?token=ot_tk_...", + "source": { + "channel": "slack", + "channelId": "my-slack", + "sender": { "id": "U456", "name": "Alice" } + }, + "message": [{ "text": "Hello from Slack!" }] +} +``` + +Access envelope fields in n8n expressions: +- `{{ $json.threadId }}` +- `{{ $json.turnId }}` +- `{{ $json.replyTo }}` +- `{{ $json.source.sender.name }}` +- `{{ $json.message[0].text }}` + +## Step 4 — Send a reply + +Add an **HTTP Request** node after your processing steps: + +| Field | Value | +|---|---| +| Method | `POST` | +| URL | `{{ $('Webhook').item.json.replyTo }}` | +| Body Content Type | `JSON` | +| Body | See below | + +**Simple text reply:** +```json +{ + "message": { "text": "Hello! I processed your message." } +} +``` + +**A2H AUTHORIZE intent (blocking — OpenThreads waits for human response):** +```json +{ + "message": [ + { "text": "I need your approval to deploy." }, + { + "intent": "AUTHORIZE", + "context": { + "action": "deploy-to-production", + "details": "Branch feature-x → production" + } + } + ] +} +``` + +## Tips + +- The `replyTo` token expires after 24 hours (configurable via `REPLY_TOKEN_TTL`). +- Store `threadId` if you need to send follow-up messages without a `replyTo`. +- Use the **Split In Batches** node to handle multiple messages in the envelope array. diff --git a/examples/plain-http/README.md b/examples/plain-http/README.md new file mode 100644 index 0000000..91ad009 --- /dev/null +++ b/examples/plain-http/README.md @@ -0,0 +1,76 @@ +# Example: Plain HTTP Webhook Consumer + +The simplest possible OpenThreads integration — a standalone HTTP server that +receives OpenThreads envelopes, processes them, and replies using `replyTo`. + +No frameworks, no SDK. Just `curl`-level HTTP. + +## How it works + +``` +Human (Slack) → OpenThreads → POST /inbound (this server) + ↓ + processes message + ↓ +Human (Slack) ← OpenThreads ← POST replyTo (this server) +``` + +## Prerequisites + +- An OpenThreads instance running at `http://localhost:3000` (see root `docker-compose.yml`) +- A channel registered in OpenThreads (Slack, Telegram, Discord, etc.) +- A route pointing to `http://localhost:4000/inbound` + +## Run the server + +```bash +bun run server.ts +# or +node server.js +``` + +The server listens on port `4000` and: +1. Receives POST requests from OpenThreads at `/inbound` +2. Echoes the message back with a text reply via `replyTo` + +## cURL test (without OpenThreads) + +You can test the inbound handler directly with curl: + +```bash +curl -s -X POST http://localhost:4000/inbound \ + -H 'Content-Type: application/json' \ + -d '{ + "threadId": "ot_thr_test", + "turnId": "ot_turn_test", + "replyTo": "http://localhost:3000/send/channel/my-slack/target/C0123/thread/ot_thr_test?token=ot_tk_test", + "source": { + "channel": "slack", + "channelId": "my-slack", + "sender": { "id": "U456", "name": "Alice" } + }, + "message": [{ "text": "Hello, agent!" }] + }' | jq . +``` + +## A2H example + +To send a human approval request back: + +```bash +# POST to replyTo with an A2H AUTHORIZE intent +curl -s -X POST "$REPLY_TO_URL" \ + -H 'Content-Type: application/json' \ + -d '{ + "message": [ + { "text": "I need your approval to proceed." }, + { + "intent": "AUTHORIZE", + "context": { + "action": "deploy-to-production", + "details": "Branch feature-x → production (12 services)" + } + } + ] + }' | jq . +``` diff --git a/examples/plain-http/server.ts b/examples/plain-http/server.ts new file mode 100644 index 0000000..d8086cd --- /dev/null +++ b/examples/plain-http/server.ts @@ -0,0 +1,120 @@ +/** + * Plain HTTP Webhook Consumer — OpenThreads example. + * + * Minimal Bun HTTP server that: + * 1. Receives POST /inbound — OpenThreads envelope (outbound from OT) + * 2. Processes the message + * 3. Replies via replyTo URL + * + * Run: bun run server.ts + */ + +const PORT = Number(process.env.PORT ?? 4000); + +// ─── Types ──────────────────────────────────────────────────────────────────── + +interface Envelope { + threadId: string; + turnId: string; + replyTo: string; + source: { + channel: string; + channelId: string; + sender: { id: string; name?: string }; + }; + message: unknown; +} + +// ─── Message processing ─────────────────────────────────────────────────────── + +async function processEnvelope(envelope: Envelope): Promise { + const { replyTo, source, message } = envelope; + console.log(`[inbound] message from ${source.sender.name ?? source.sender.id} on ${source.channel}:`, message); + + // Build a reply + const reply = buildReply(message, source.sender.name ?? source.sender.id); + + // Send the reply via replyTo + const response = await fetch(replyTo, { + method: 'POST', + headers: { 'Content-Type': 'application/json' }, + body: JSON.stringify(reply), + }); + + if (!response.ok) { + const body = await response.text().catch(() => ''); + console.error(`[reply] failed: ${response.status} ${body}`); + } else { + console.log(`[reply] sent to ${replyTo}`); + } +} + +function buildReply(message: unknown, senderName: string): { message: unknown } { + // Extract text from the message + let text = ''; + if (Array.isArray(message)) { + const texts = message + .filter((m): m is { text: string } => typeof m === 'object' && m !== null && 'text' in m) + .map((m) => m.text); + text = texts.join(' '); + } else if (typeof message === 'object' && message !== null && 'text' in message) { + text = (message as { text: string }).text; + } + + // Echo the message back + return { + message: { + text: `Echo from webhook consumer: "${text}" (from ${senderName})`, + }, + }; +} + +// ─── HTTP server ────────────────────────────────────────────────────────────── + +const server = Bun.serve({ + port: PORT, + async fetch(req) { + const url = new URL(req.url); + + // POST /inbound — receive OpenThreads envelope + if (req.method === 'POST' && url.pathname === '/inbound') { + let body: unknown; + try { + body = await req.json(); + } catch { + return new Response(JSON.stringify({ error: 'Invalid JSON' }), { + status: 400, + headers: { 'Content-Type': 'application/json' }, + }); + } + + const envelope = body as Envelope; + + // Acknowledge immediately, process asynchronously + void processEnvelope(envelope).catch((err: unknown) => { + console.error('[processEnvelope] error:', err); + }); + + return new Response(JSON.stringify({ ok: true }), { + status: 200, + headers: { 'Content-Type': 'application/json' }, + }); + } + + // GET /health — liveness probe + if (req.method === 'GET' && url.pathname === '/health') { + return new Response(JSON.stringify({ status: 'ok' }), { + headers: { 'Content-Type': 'application/json' }, + }); + } + + return new Response(JSON.stringify({ error: 'Not found' }), { + status: 404, + headers: { 'Content-Type': 'application/json' }, + }); + }, +}); + +console.log(`[server] listening on http://localhost:${server.port}`); +console.log(`[server] POST /inbound — receive OpenThreads envelopes`); +console.log(`[server] GET /health — liveness probe`); diff --git a/package.json b/package.json index e3ca2c1..fecad18 100644 --- a/package.json +++ b/package.json @@ -7,7 +7,8 @@ "packages/storage/*", "packages/channels/*", "packages/server", - "packages/trust" + "packages/trust", + "packages/create-openthreads" ], "scripts": { "test": "bun test", diff --git a/packages/core/src/index.ts b/packages/core/src/index.ts index 3fbdbe6..96ffcba 100644 --- a/packages/core/src/index.ts +++ b/packages/core/src/index.ts @@ -17,7 +17,20 @@ export type { } from './types/envelope.js'; // ---- A2H Protocol Types ---- -export type { A2HMessage, A2HIntent, A2HContext } from './types/a2h.js'; +export type { + A2HMessage, + A2HIntent, + A2HContext, + // Layer 1 (implemented) + A2HLayer1Intent, + // Layer 2 stubs — pending twilio-labs/a2h-spec stabilisation + A2HLayer2Intent, + A2HPolicyIntent, + A2HRevokeIntent, + A2HDelegateIntent, + A2HScopeIntent, +} from './types/a2h.js'; +export { isLayer1Intent, isLayer2Intent } from './types/a2h.js'; // ---- Message Union Types ---- export type { @@ -99,3 +112,14 @@ export type { ThreadManagerOptions } from './thread/index.js'; export { TurnManager } from './turn/index.js'; export type { TurnManagerOptions } from './turn/index.js'; + +// ---- Reply Engine Extension Point ---- +export { + intentHandlerRegistry, + IntentHandlerRegistry, +} from './reply-engine/intent-handler.js'; +export type { + IntentHandlerFn, + IntentHandlerContext, + IntentHandlerResponse, +} from './reply-engine/intent-handler.js'; diff --git a/packages/core/src/reply-engine/intent-handler.ts b/packages/core/src/reply-engine/intent-handler.ts new file mode 100644 index 0000000..96fe8ae --- /dev/null +++ b/packages/core/src/reply-engine/intent-handler.ts @@ -0,0 +1,133 @@ +/** + * Extension point for custom A2H intent handlers in the Reply Engine. + * + * This module defines the IntentHandler interface and the IntentHandlerRegistry, + * which allows registering handlers for Layer 2 intents (POLICY, REVOKE, DELEGATE, + * SCOPE) or overriding the default handling of Layer 1 intents. + * + * ## When to use + * + * - **Layer 2 intents (stub):** Register handlers for `POLICY`, `REVOKE`, `DELEGATE`, + * `SCOPE` when the A2H Layer 2 spec stabilises. Until then, unhandled Layer 2 intents + * are acknowledged with a `layer2_not_implemented` response. + * + * - **Custom Layer 1 overrides:** Override the built-in handling for `INFORM`, `COLLECT`, + * `AUTHORIZE`, `ESCALATE`, or `RESULT` for specific deployments (e.g., custom auth flow). + * + * ## Registration + * + * ```ts + * import { intentHandlerRegistry } from '@openthreads/core/reply-engine'; + * + * intentHandlerRegistry.register('POLICY', async (message, context) => { + * // Parse the standing approval rule from message.context + * // Persist to your policy store + * // Return an acknowledgement + * return { + * intent: 'POLICY', + * response: { acknowledged: true, policyId: 'pol_...' }, + * respondedAt: new Date(), + * }; + * }); + * ``` + * + * @see https://github.com/twilio-labs/a2h-spec — Layer 2 spec (roadmap) + */ + +import type { A2HMessage, A2HIntent } from '../types/a2h.js'; + +// ─── IntentHandler interface ────────────────────────────────────────────────── + +/** + * The context passed to a custom intent handler. + */ +export interface IntentHandlerContext { + /** The turn identifier for this interaction. */ + turnId: string; + /** The OpenThreads channel ID this intent arrived on. */ + channelId?: string; +} + +/** + * The response returned by a custom intent handler. + * Mirrors the A2HResponse shape expected by the Reply Engine. + */ +export interface IntentHandlerResponse { + intent: A2HIntent; + /** The handler's response payload. Shape is intent-specific. */ + response: unknown; + /** Optional human-readable comment (e.g., for AUTHORIZE). */ + comment?: string; + /** When the response was generated. */ + respondedAt: Date; + /** Handler-specific metadata (for audit/debug). */ + meta?: Record; +} + +/** + * A handler function for a specific A2H intent type. + * + * Handlers are async functions that receive an A2H message and return a response. + * Throwing from a handler causes the Reply Engine to reject the intent with an error. + */ +export type IntentHandlerFn = ( + message: A2HMessage, + context: IntentHandlerContext, +) => Promise; + +// ─── Registry ───────────────────────────────────────────────────────────────── + +/** + * Registry of custom intent handlers. + * + * Handlers registered here are invoked by the Reply Engine when it encounters + * an intent of the matching type. Built-in Layer 1 handling takes precedence + * unless a handler is explicitly registered for a Layer 1 intent type. + */ +export class IntentHandlerRegistry { + private readonly handlers = new Map(); + + /** + * Register a handler for the given intent type. + * Replaces any previously registered handler for the same intent. + */ + register(intent: A2HIntent, handler: IntentHandlerFn): this { + this.handlers.set(intent, handler); + return this; + } + + /** + * Remove the handler for the given intent type. + */ + unregister(intent: A2HIntent): this { + this.handlers.delete(intent); + return this; + } + + /** + * Returns the handler registered for the given intent type, or `undefined` if none. + */ + get(intent: A2HIntent): IntentHandlerFn | undefined { + return this.handlers.get(intent); + } + + /** + * Returns true if a handler is registered for the given intent type. + */ + has(intent: A2HIntent): boolean { + return this.handlers.has(intent); + } +} + +/** + * Global intent handler registry shared across all Reply Engine instances. + * + * Register Layer 2 handlers here once the A2H spec stabilises: + * ```ts + * import { intentHandlerRegistry } from '@openthreads/core'; + * + * intentHandlerRegistry.register('POLICY', myPolicyHandler); + * intentHandlerRegistry.register('REVOKE', myRevokeHandler); + * ``` + */ +export const intentHandlerRegistry = new IntentHandlerRegistry(); diff --git a/packages/core/src/types/a2h.ts b/packages/core/src/types/a2h.ts index a7fb4a8..0b4e094 100644 --- a/packages/core/src/types/a2h.ts +++ b/packages/core/src/types/a2h.ts @@ -1,6 +1,7 @@ /** * A2H (Agent-to-Human) Protocol intent types. * + * ── Layer 1 (implemented) ─────────────────────────────────────────────────── * 5 atomic, composable intents from the A2H spec (Twilio, Feb 2026): * * - INFORM: Fire-and-forget notification. "Letting you know I did X." @@ -8,8 +9,53 @@ * - AUTHORIZE: Blocking request for approval with evidence. "Can I deploy to prod?" * - ESCALATE: Handoff to a human operator. * - RESULT: Returns a task result to the agent. + * + * ── Layer 2 (stubs — pending spec stabilisation) ─────────────────────────── + * Autonomy governance intents from the A2H roadmap. These are stubbed here to + * allow type-safe extension without breaking Layer 1 consumers. Monitor + * https://github.com/twilio-labs/a2h-spec for Layer 2 stabilisation. + * + * - POLICY: Define a standing approval rule. "Pre-approve all deploys under $100." + * - REVOKE: Cancel a previously granted policy or delegation. + * - DELEGATE: Grant another agent or human the ability to act on your behalf. + * - SCOPE: Restrict or expand the authority of an agent for a defined context. + */ + +// ── Layer 1 intents ─────────────────────────────────────────────────────────── + +export type A2HLayer1Intent = 'INFORM' | 'COLLECT' | 'AUTHORIZE' | 'ESCALATE' | 'RESULT'; + +// ── Layer 2 intents (stubs — not yet processed by the Reply Engine) ─────────── + +/** + * POLICY — define a standing approval rule (e.g. "approve all transactions < $50"). + * @stub Layer 2 — monitor twilio-labs/a2h-spec for spec stabilisation. + */ +export type A2HPolicyIntent = 'POLICY'; + +/** + * REVOKE — cancel a previously granted policy or delegation. + * @stub Layer 2 — monitor twilio-labs/a2h-spec for spec stabilisation. + */ +export type A2HRevokeIntent = 'REVOKE'; + +/** + * DELEGATE — grant another principal (agent or human) the ability to act on behalf of this principal. + * @stub Layer 2 — monitor twilio-labs/a2h-spec for spec stabilisation. */ -export type A2HIntent = 'INFORM' | 'COLLECT' | 'AUTHORIZE' | 'ESCALATE' | 'RESULT'; +export type A2HDelegateIntent = 'DELEGATE'; + +/** + * SCOPE — restrict or expand an agent's authority for a defined context window. + * @stub Layer 2 — monitor twilio-labs/a2h-spec for spec stabilisation. + */ +export type A2HScopeIntent = 'SCOPE'; + +/** All Layer 2 intents (stubs). */ +export type A2HLayer2Intent = A2HPolicyIntent | A2HRevokeIntent | A2HDelegateIntent | A2HScopeIntent; + +/** Union of all A2H intents (Layer 1 + Layer 2 stubs). */ +export type A2HIntent = A2HLayer1Intent | A2HLayer2Intent; /** * Context payload carried within an A2H message. Contains structured @@ -33,3 +79,23 @@ export interface A2HMessage { /** Idempotency key to prevent duplicate processing */ idempotencyKey?: string; } + +// ─── Type guards ────────────────────────────────────────────────────────────── + +const LAYER1_INTENTS: ReadonlySet = new Set([ + 'INFORM', 'COLLECT', 'AUTHORIZE', 'ESCALATE', 'RESULT', +]); + +const LAYER2_INTENTS: ReadonlySet = new Set([ + 'POLICY', 'REVOKE', 'DELEGATE', 'SCOPE', +]); + +/** Returns true if the intent is a Layer 1 intent (fully supported). */ +export function isLayer1Intent(intent: A2HIntent): intent is A2HLayer1Intent { + return LAYER1_INTENTS.has(intent); +} + +/** Returns true if the intent is a Layer 2 stub (not yet processed). */ +export function isLayer2Intent(intent: A2HIntent): intent is A2HLayer2Intent { + return LAYER2_INTENTS.has(intent); +} diff --git a/packages/create-openthreads/package.json b/packages/create-openthreads/package.json new file mode 100644 index 0000000..560e1b1 --- /dev/null +++ b/packages/create-openthreads/package.json @@ -0,0 +1,18 @@ +{ + "name": "create-openthreads", + "version": "0.1.0", + "description": "Scaffold a new OpenThreads deployment", + "type": "module", + "bin": { + "create-openthreads": "./src/index.ts" + }, + "scripts": { + "start": "bun src/index.ts" + }, + "devDependencies": { + "@types/bun": "latest", + "typescript": "^5.0.0" + }, + "keywords": ["openthreads", "scaffold", "create"], + "license": "MIT" +} diff --git a/packages/create-openthreads/src/index.ts b/packages/create-openthreads/src/index.ts new file mode 100644 index 0000000..96b3c3d --- /dev/null +++ b/packages/create-openthreads/src/index.ts @@ -0,0 +1,180 @@ +#!/usr/bin/env bun +/** + * create-openthreads — scaffold a new OpenThreads deployment. + * + * Usage: + * bunx create-openthreads → interactive mode + * bunx create-openthreads my-app → create in ./my-app + * npx create-openthreads my-app → same, via npm + */ + +import { existsSync, mkdirSync, writeFileSync } from 'fs'; +import { join } from 'path'; + +const OPENTHREADS_VERSION = '0.1.0'; + +// ─── CLI args ───────────────────────────────────────────────────────────────── + +const args = process.argv.slice(2); +let projectName = args[0]; + +if (!projectName) { + process.stdout.write('Project name: '); + // Read from stdin synchronously (Bun supports this) + const buf = Buffer.alloc(256); + const n = require('fs').readSync(0, buf, 0, buf.length, null); + projectName = buf.toString('utf8', 0, n).trim(); +} + +if (!projectName || projectName.startsWith('-')) { + console.error('Usage: create-openthreads '); + process.exit(1); +} + +const targetDir = join(process.cwd(), projectName); + +if (existsSync(targetDir)) { + console.error(`Directory "${projectName}" already exists.`); + process.exit(1); +} + +// ─── Scaffold ───────────────────────────────────────────────────────────────── + +console.log(`\nCreating OpenThreads project: ${projectName}\n`); + +mkdirSync(targetDir, { recursive: true }); + +function write(relativePath: string, content: string): void { + const fullPath = join(targetDir, relativePath); + const dir = fullPath.substring(0, fullPath.lastIndexOf('/')); + if (dir) mkdirSync(dir, { recursive: true }); + writeFileSync(fullPath, content, 'utf8'); + console.log(` created ${relativePath}`); +} + +// docker-compose.yml +write('docker-compose.yml', `version: '3.8' + +services: + mongodb: + image: mongo:7.0 + container_name: ${projectName}-mongodb + ports: + - '27017:27017' + environment: + MONGO_INITDB_ROOT_USERNAME: openthreads + MONGO_INITDB_ROOT_PASSWORD: openthreads + MONGO_INITDB_DATABASE: openthreads + volumes: + - mongodb_data:/data/db + healthcheck: + test: ['CMD', 'mongosh', '--eval', "db.adminCommand('ping')"] + interval: 10s + timeout: 5s + retries: 5 + start_period: 10s + restart: unless-stopped + + app: + profiles: [production] + image: ghcr.io/deepducks/openthreads:latest + container_name: ${projectName}-app + ports: + - '\${PORT:-3000}:3000' + environment: + NODE_ENV: production + MONGODB_URI: mongodb://openthreads:openthreads@mongodb:27017/openthreads + JWT_SECRET: \${JWT_SECRET} + OPENTHREADS_BASE_URL: \${OPENTHREADS_BASE_URL:-http://localhost:3000} + LOG_LEVEL: \${LOG_LEVEL:-info} + LOG_FORMAT: json + depends_on: + mongodb: + condition: service_healthy + restart: unless-stopped + +volumes: + mongodb_data: + driver: local +`); + +// .env +write('.env', `# OpenThreads configuration +# Copy this to .env and fill in your values. + +PORT=3000 +NODE_ENV=development + +# MongoDB +MONGODB_URI=mongodb://openthreads:openthreads@localhost:27017/openthreads + +# Security — CHANGE THESE IN PRODUCTION +JWT_SECRET=change-me-in-production +# MANAGEMENT_API_KEY=change-me-in-production + +# Base URL (used to build replyTo URLs) +OPENTHREADS_BASE_URL=http://localhost:3000 + +# Logging +LOG_LEVEL=info +LOG_FORMAT=text + +# Reply token TTL (seconds) +REPLY_TOKEN_TTL=86400 + +# Channel credentials (uncomment the ones you need) +# SLACK_BOT_TOKEN=xoxb-... +# SLACK_SIGNING_SECRET=... +# TELEGRAM_BOT_TOKEN=... +# DISCORD_BOT_TOKEN=... +# DISCORD_CLIENT_ID=... +`); + +// .gitignore +write('.gitignore', `.env +.env.local +node_modules/ +.next/ +*.log +`); + +// README.md +write('README.md', `# ${projectName} + +OpenThreads deployment scaffolded with \`create-openthreads\`. + +## Quick start + +1. **Start MongoDB:** + \`\`\`bash + docker compose up -d mongodb + \`\`\` + +2. **Configure environment:** + \`\`\`bash + cp .env.example .env + # Edit .env with your channel credentials and secrets + \`\`\` + +3. **Start the server:** + \`\`\`bash + docker compose --profile production up -d + # or for development: cd into OpenThreads and run bun run dev + \`\`\` + +4. **Open the dashboard:** + [http://localhost:3000](http://localhost:3000) + +## Documentation + +- [Self-hosting guide](https://github.com/deepducks/OpenThreads/blob/main/docs/self-hosting.md) +- [Channel setup guides](https://github.com/deepducks/OpenThreads/blob/main/docs/channels/) +- [API reference](http://localhost:3000/api/docs) +`); + +console.log(`\nDone! Next steps:\n`); +console.log(` cd ${projectName}`); +console.log(` cp .env .env.local # edit with your credentials`); +console.log(` docker compose up -d mongodb`); +console.log(` docker compose --profile production up -d`); +console.log(`\n Dashboard: http://localhost:3000\n`); diff --git a/packages/server/src/app/api/metrics/route.ts b/packages/server/src/app/api/metrics/route.ts new file mode 100644 index 0000000..949b0d6 --- /dev/null +++ b/packages/server/src/app/api/metrics/route.ts @@ -0,0 +1,43 @@ +/** + * GET /api/metrics — Prometheus-compatible metrics endpoint. + * + * Returns metrics in the Prometheus text exposition format (Content-Type: + * text/plain; version=0.0.4). Scrape this endpoint from your Prometheus + * configuration: + * + * scrape_configs: + * - job_name: openthreads + * static_configs: + * - targets: ['openthreads:3000'] + * metrics_path: /api/metrics + * + * Access can be restricted by setting MANAGEMENT_API_KEY in the environment. + */ + +import { NextRequest, NextResponse } from 'next/server'; +import { registry } from '@/lib/metrics'; + +export const runtime = 'nodejs'; +// Prevent Next.js from caching this route — metrics must always be fresh. +export const dynamic = 'force-dynamic'; + +export async function GET(request: NextRequest): Promise { + // If a management API key is configured, require it on the metrics endpoint too. + const apiKey = process.env.MANAGEMENT_API_KEY; + if (apiKey) { + const auth = request.headers.get('authorization') ?? ''; + const token = auth.startsWith('Bearer ') ? auth.slice(7) : ''; + if (token !== apiKey) { + return new NextResponse('Unauthorized', { status: 401 }); + } + } + + const body = registry.render(); + + return new NextResponse(body, { + status: 200, + headers: { + 'Content-Type': 'text/plain; version=0.0.4; charset=utf-8', + }, + }); +} diff --git a/packages/server/src/instrumentation.ts b/packages/server/src/instrumentation.ts index 709532b..9916521 100644 --- a/packages/server/src/instrumentation.ts +++ b/packages/server/src/instrumentation.ts @@ -1,29 +1,91 @@ /** * Next.js instrumentation file. * - * Called once when the server starts. Used to set up graceful shutdown - * handling so that active MongoDB connections are closed cleanly when - * the process receives SIGTERM or SIGINT. + * Called once when the server starts. Responsibilities: + * 1. Graceful shutdown — close MongoDB connections cleanly on SIGTERM/SIGINT. + * 2. OpenTelemetry — initialise tracing when OTEL_EXPORTER_OTLP_ENDPOINT is set. + * + * OpenTelemetry configuration (via environment variables): + * + * OTEL_EXPORTER_OTLP_ENDPOINT OTLP gRPC/HTTP endpoint, e.g. http://otel-collector:4318 + * OTEL_SERVICE_NAME Service name tag (default: "openthreads") + * OTEL_SDK_DISABLED Set to "true" to disable tracing entirely * * @see https://nextjs.org/docs/app/building-your-application/optimizing/instrumentation + * @see https://opentelemetry.io/docs/languages/js/getting-started/nodejs/ */ export async function register(): Promise { - if (process.env.NEXT_RUNTIME === 'nodejs') { - const { disconnectDb } = await import('./lib/db'); - - async function shutdown(signal: string): Promise { - console.log(`[shutdown] received ${signal}, closing connections…`); - try { - await disconnectDb(); - console.log('[shutdown] MongoDB connection closed'); - } catch (err) { - console.error('[shutdown] error closing MongoDB connection:', err); - } - process.exit(0); + if (process.env.NEXT_RUNTIME !== 'nodejs') return; + + // ── Graceful shutdown ──────────────────────────────────────────────────────── + const { disconnectDb } = await import('./lib/db'); + + async function shutdown(signal: string): Promise { + console.log(JSON.stringify({ level: 'info', message: `received ${signal}, shutting down…`, signal })); + try { + await disconnectDb(); + console.log(JSON.stringify({ level: 'info', message: 'MongoDB connection closed' })); + } catch (err) { + console.error(JSON.stringify({ + level: 'error', + message: 'error closing MongoDB connection', + error: err instanceof Error ? err.message : String(err), + })); } + process.exit(0); + } + + process.on('SIGTERM', () => { void shutdown('SIGTERM'); }); + process.on('SIGINT', () => { void shutdown('SIGINT'); }); + + // ── OpenTelemetry tracing ─────────────────────────────────────────────────── + if ( + process.env.OTEL_SDK_DISABLED !== 'true' && + process.env.OTEL_EXPORTER_OTLP_ENDPOINT + ) { + await setupOpenTelemetry(); + } +} + +async function setupOpenTelemetry(): Promise { + try { + const { + NodeSDK, + // eslint-disable-next-line @typescript-eslint/no-require-imports + } = await import('@opentelemetry/sdk-node' as string).catch(() => ({ NodeSDK: null })) as { NodeSDK: (new (...args: unknown[]) => { start(): void }) | null }; + + if (!NodeSDK) { + console.warn(JSON.stringify({ + level: 'warn', + message: 'OpenTelemetry SDK not installed. Install @opentelemetry/sdk-node to enable tracing.', + })); + return; + } + + const serviceName = process.env.OTEL_SERVICE_NAME ?? 'openthreads'; + + // NodeSDK auto-instruments HTTP, DNS, MongoDB, etc. via OTEL_NODE_RESOURCE_DETECTORS. + // The exporter endpoint and protocol are read from OTEL_EXPORTER_OTLP_ENDPOINT / + // OTEL_EXPORTER_OTLP_PROTOCOL (defaults to http/protobuf). + const sdk = new NodeSDK({ resource: { attributes: { 'service.name': serviceName } } } as unknown as Parameters[0]); + sdk.start(); + + console.log(JSON.stringify({ + level: 'info', + message: 'OpenTelemetry tracing started', + serviceName, + endpoint: process.env.OTEL_EXPORTER_OTLP_ENDPOINT, + })); - process.on('SIGTERM', () => { void shutdown('SIGTERM'); }); - process.on('SIGINT', () => { void shutdown('SIGINT'); }); + // Flush spans on shutdown + process.on('SIGTERM', () => { void (sdk as unknown as { shutdown(): Promise }).shutdown(); }); + process.on('SIGINT', () => { void (sdk as unknown as { shutdown(): Promise }).shutdown(); }); + } catch (err) { + console.error(JSON.stringify({ + level: 'error', + message: 'Failed to start OpenTelemetry SDK', + error: err instanceof Error ? err.message : String(err), + })); } } diff --git a/packages/server/src/lib/logger.ts b/packages/server/src/lib/logger.ts index d2a0097..dd44574 100644 --- a/packages/server/src/lib/logger.ts +++ b/packages/server/src/lib/logger.ts @@ -1,10 +1,78 @@ /** - * Minimal request logger for OpenThreads server. + * Structured JSON logger for OpenThreads server. * - * Logs method, path, status, and duration in a structured format. - * In production, plug in your preferred logger (Pino, Winston, etc.). + * Outputs newline-delimited JSON when LOG_FORMAT=json (default in production) + * or human-readable text when LOG_FORMAT=text (default in development). + * + * Log level is controlled via the LOG_LEVEL environment variable: + * debug | info | warn | error (default: info) */ +export type LogLevel = 'debug' | 'info' | 'warn' | 'error'; + +const LEVEL_RANK: Record = { + debug: 0, + info: 1, + warn: 2, + error: 3, +}; + +function getConfiguredLevel(): LogLevel { + const env = (process.env.LOG_LEVEL ?? 'info').toLowerCase() as LogLevel; + return env in LEVEL_RANK ? env : 'info'; +} + +function isJsonFormat(): boolean { + const fmt = process.env.LOG_FORMAT ?? (process.env.NODE_ENV === 'production' ? 'json' : 'text'); + return fmt === 'json'; +} + +function shouldLog(level: LogLevel): boolean { + return LEVEL_RANK[level] >= LEVEL_RANK[getConfiguredLevel()]; +} + +// ─── Core log function ──────────────────────────────────────────────────────── + +function log(level: LogLevel, message: string, fields?: Record): void { + if (!shouldLog(level)) return; + + if (isJsonFormat()) { + const entry: Record = { + timestamp: new Date().toISOString(), + level, + message, + ...fields, + }; + const line = JSON.stringify(entry); + if (level === 'error') { + process.stderr.write(line + '\n'); + } else { + process.stdout.write(line + '\n'); + } + } else { + const ts = new Date().toISOString(); + const lvl = level.toUpperCase().padEnd(5); + const extras = fields ? ' ' + Object.entries(fields).map(([k, v]) => `${k}=${String(v)}`).join(' ') : ''; + const line = `[${ts}] [${lvl}] ${message}${extras}`; + if (level === 'error') { + console.error(line); + } else if (level === 'warn') { + console.warn(line); + } else { + console.log(line); + } + } +} + +export const logger = { + debug: (message: string, fields?: Record) => log('debug', message, fields), + info: (message: string, fields?: Record) => log('info', message, fields), + warn: (message: string, fields?: Record) => log('warn', message, fields), + error: (message: string, fields?: Record) => log('error', message, fields), +}; + +// ─── Request logging ────────────────────────────────────────────────────────── + export interface RequestLogEntry { timestamp: string; method: string; @@ -16,31 +84,21 @@ export interface RequestLogEntry { } export function logRequest(entry: RequestLogEntry): void { - const { timestamp, method, path, status, durationMs, ip, error } = entry; - const level = status >= 500 ? 'ERROR' : status >= 400 ? 'WARN' : 'INFO'; - const parts = [ - `[${timestamp}]`, - `[${level}]`, + const { status, method, path, durationMs, ip, error } = entry; + const level: LogLevel = status >= 500 ? 'error' : status >= 400 ? 'warn' : 'info'; + + log(level, `${method} ${path} ${status}`, { method, path, status, - `${durationMs}ms`, - ]; - if (ip) parts.push(`ip=${ip}`); - if (error) parts.push(`error=${error}`); - - if (level === 'ERROR') { - console.error(parts.join(' ')); - } else if (level === 'WARN') { - console.warn(parts.join(' ')); - } else { - console.log(parts.join(' ')); - } + durationMs, + ...(ip ? { ip } : {}), + ...(error ? { error } : {}), + }); } /** * Create a request log entry from a Next.js Request + Response. - * Call this at the end of a route handler, passing the start time. */ export function createLogEntry( request: Request, diff --git a/packages/server/src/lib/metrics.ts b/packages/server/src/lib/metrics.ts new file mode 100644 index 0000000..42febc7 --- /dev/null +++ b/packages/server/src/lib/metrics.ts @@ -0,0 +1,141 @@ +/** + * In-process metrics registry for OpenThreads. + * + * Tracks Prometheus-compatible counters and gauges. Values are exposed via + * GET /api/metrics in the Prometheus text exposition format. + * + * All counters and gauges are process-local. In a multi-instance deployment, + * aggregate across instances using a Prometheus scrape job. + */ + +type MetricType = 'counter' | 'gauge'; + +interface MetricDescriptor { + name: string; + help: string; + type: MetricType; +} + +interface LabelSet { + [label: string]: string; +} + +interface Sample { + labels: LabelSet; + value: number; +} + +class Metric { + private samples = new Map(); + + constructor(readonly descriptor: MetricDescriptor) {} + + private key(labels: LabelSet): string { + return JSON.stringify(Object.fromEntries(Object.entries(labels).sort())); + } + + inc(labels: LabelSet = {}, amount = 1): void { + const k = this.key(labels); + const existing = this.samples.get(k); + if (existing) { + existing.value += amount; + } else { + this.samples.set(k, { labels, value: amount }); + } + } + + set(labels: LabelSet = {}, value: number): void { + const k = this.key(labels); + this.samples.set(k, { labels, value }); + } + + get(labels: LabelSet = {}): number { + return this.samples.get(this.key(labels))?.value ?? 0; + } + + render(): string { + const { name, help, type } = this.descriptor; + const lines: string[] = [ + `# HELP ${name} ${help}`, + `# TYPE ${name} ${type}`, + ]; + for (const { labels, value } of this.samples.values()) { + const labelStr = Object.entries(labels) + .map(([k, v]) => `${k}="${v.replace(/"/g, '\\"')}"`) + .join(','); + lines.push(labelStr ? `${name}{${labelStr}} ${value}` : `${name} ${value}`); + } + return lines.join('\n'); + } +} + +// ─── Registry ───────────────────────────────────────────────────────────────── + +class MetricsRegistry { + private metrics = new Map(); + + private register(descriptor: MetricDescriptor): Metric { + if (!this.metrics.has(descriptor.name)) { + this.metrics.set(descriptor.name, new Metric(descriptor)); + } + return this.metrics.get(descriptor.name)!; + } + + counter(name: string, help: string): Metric { + return this.register({ name, help, type: 'counter' }); + } + + gauge(name: string, help: string): Metric { + return this.register({ name, help, type: 'gauge' }); + } + + render(): string { + return [...this.metrics.values()].map((m) => m.render()).join('\n\n') + '\n'; + } +} + +export const registry = new MetricsRegistry(); + +// ─── Metric definitions ─────────────────────────────────────────────────────── + +/** Total inbound webhook events received, labelled by channel. */ +export const messagesInTotal = registry.counter( + 'openthreads_messages_in_total', + 'Total number of inbound messages received from channels.', +); + +/** Total outbound messages sent to recipients, labelled by channel and status. */ +export const messagesOutTotal = registry.counter( + 'openthreads_messages_out_total', + 'Total number of outbound messages sent to recipients.', +); + +/** Total A2H intents processed, labelled by intent type and method (1-4). */ +export const a2hIntentsTotal = registry.counter( + 'openthreads_a2h_intents_total', + 'Total number of A2H intents processed by the Reply Engine.', +); + +/** Number of currently active (open, not-yet-resolved) threads. */ +export const activeThreadsGauge = registry.gauge( + 'openthreads_active_threads', + 'Number of active (open) threads.', +); + +/** HTTP request duration histogram approximation (p50/p95 via labelled gauges). */ +export const httpRequestDurationMs = registry.counter( + 'openthreads_http_requests_total', + 'Total HTTP requests served, labelled by method, path, and status_class.', +); + +/** Recipient fanout latency total (for computing average). */ +export const fanoutDurationMsTotal = registry.counter( + 'openthreads_fanout_duration_ms_total', + 'Cumulative fanout latency in milliseconds (divide by fanout_count for average).', +); + +/** Total successful fanout deliveries. */ +export const fanoutTotal = registry.counter( + 'openthreads_fanout_total', + 'Total recipient fanout attempts, labelled by status (success|error).', +);