From 51fe46fd36548e53d0f2a0ed0b41d25095cc5a79 Mon Sep 17 00:00:00 2001 From: sirily11 <32106111+sirily11@users.noreply.github.com> Date: Wed, 20 May 2026 13:28:49 +0800 Subject: [PATCH] feat: enable the k8s deployment for the relay server --- .github/workflows/relay-deploy.yaml | 140 ++++++++++++++ .gitignore | 1 + .../RxCodeChatKit/ChatMessageBubble.swift | 6 +- RxCode/App/AppState.swift | 11 +- RxCode/Services/ClaudeService.swift | 36 ++++ RxCode/Services/MCPService.swift | 19 +- RxCode/Services/RelayPresetCatalog.swift | 72 ++++++++ RxCode/Views/Settings/MobileSettingsTab.swift | 171 ++++++++++++++++-- relay-server/Dockerfile | 5 +- relay-server/README.md | 32 ++++ relay-server/backplane.go | 139 ++++++++++++++ relay-server/go.mod | 5 +- relay-server/go.sum | 19 +- relay-server/health.go | 7 +- relay-server/k8s/README.md | 84 +++++++++ relay-server/k8s/configmap.yaml | 17 ++ relay-server/k8s/deployment.yaml | 68 +++++++ relay-server/k8s/hpa.yaml | 26 +++ relay-server/k8s/ingress.yaml | 34 ++++ relay-server/k8s/kustomization.yaml | 23 +++ relay-server/k8s/namespace.yaml | 8 + relay-server/k8s/pdb.yaml | 14 ++ relay-server/k8s/service.yaml | 16 ++ relay-server/main.go | 17 ++ relay-server/relay.go | 56 +++++- website/.gitignore | 1 + website/public/relays.json | 13 ++ 27 files changed, 1016 insertions(+), 24 deletions(-) create mode 100644 .github/workflows/relay-deploy.yaml create mode 100644 RxCode/Services/RelayPresetCatalog.swift create mode 100644 relay-server/backplane.go create mode 100644 relay-server/k8s/README.md create mode 100644 relay-server/k8s/configmap.yaml create mode 100644 relay-server/k8s/deployment.yaml create mode 100644 relay-server/k8s/hpa.yaml create mode 100644 relay-server/k8s/ingress.yaml create mode 100644 relay-server/k8s/kustomization.yaml create mode 100644 relay-server/k8s/namespace.yaml create mode 100644 relay-server/k8s/pdb.yaml create mode 100644 relay-server/k8s/service.yaml create mode 100644 website/public/relays.json diff --git a/.github/workflows/relay-deploy.yaml b/.github/workflows/relay-deploy.yaml new file mode 100644 index 00000000..9eb02903 --- /dev/null +++ b/.github/workflows/relay-deploy.yaml @@ -0,0 +1,140 @@ +name: Deploy Relay Server + +concurrency: + group: ${{ github.workflow }}-${{ github.ref }} + cancel-in-progress: true + +on: + push: + branches: + - main + release: + types: + - created + +env: + IMAGE: ghcr.io/${{ github.repository_owner }}/rxcode-relay + +jobs: + build: + name: Build Relay Image + runs-on: ubuntu-latest + permissions: + contents: read + packages: write + steps: + - name: Checkout repository + uses: actions/checkout@v6 + + - name: Set up Docker Buildx + uses: docker/setup-buildx-action@v4 + + - name: Log in to GitHub Container Registry + if: github.event_name == 'release' + uses: docker/login-action@v4 + with: + registry: ghcr.io + username: ${{ github.actor }} + password: ${{ secrets.GITHUB_TOKEN }} + + - name: Extract metadata + id: meta + uses: docker/metadata-action@v6 + with: + images: ${{ env.IMAGE }} + tags: | + type=semver,pattern=v{{version}} + type=semver,pattern=v{{major}}.{{minor}} + type=raw,value=latest,enable=${{ github.event_name == 'release' }} + type=sha,prefix=sha- + + # Push / pull-request: build only, then smoke-test the binary boots in + # single-node mode (no REDIS_URL) and serves /healthz. No image is + # pushed and no deploy happens — that is release-gated below. + - name: Build Docker image (test only) + if: github.event_name != 'release' + uses: docker/build-push-action@v7 + with: + context: relay-server + push: false + load: true + tags: rxcode-relay:ci + cache-from: type=gha,scope=rxcode-relay + cache-to: type=gha,mode=max,scope=rxcode-relay + + - name: Health check (single-node) + if: github.event_name != 'release' + run: | + docker run -d --name relay-test -p 8787:8787 rxcode-relay:ci + for i in $(seq 1 30); do + if curl -sf http://localhost:8787/healthz > /dev/null 2>&1; then + echo "Health check passed" + curl -s http://localhost:8787/healthz + docker rm -f relay-test + exit 0 + fi + sleep 1 + done + echo "Health check failed" + docker logs relay-test + docker rm -f relay-test + exit 1 + + - name: Build and push Docker image + if: github.event_name == 'release' + uses: docker/build-push-action@v7 + with: + context: relay-server + platforms: linux/amd64 + push: true + tags: ${{ steps.meta.outputs.tags }} + labels: ${{ steps.meta.outputs.labels }} + cache-from: type=gha,scope=rxcode-relay + cache-to: type=gha,mode=max,scope=rxcode-relay + + deploy: + name: Deploy to Kubernetes + needs: build + runs-on: ubuntu-latest + if: github.event_name == 'release' + permissions: + contents: read + steps: + - name: Checkout repository + uses: actions/checkout@v6 + + - name: Set up kubectl + uses: azure/setup-kubectl@v5 + with: + version: "latest" + + - name: Configure kubectl + run: | + mkdir -p ~/.kube + echo "${{ secrets.K8S_CONFIG_FILE_B64 }}" | base64 -d > ~/.kube/config + chmod 600 ~/.kube/config + + - name: Verify kubectl connection + run: kubectl cluster-info + + - name: Deploy with kustomize + run: kubectl kustomize relay-server/k8s/ | kubectl apply -f - + + - name: Pin deployment to this release + run: | + VERSION="${{ github.ref_name }}" + kubectl set image deployment/rxcode-relay \ + relay=${{ env.IMAGE }}:${VERSION} \ + -n rxcode-relay + + - name: Wait for rollout + run: kubectl rollout status deployment/rxcode-relay -n rxcode-relay --timeout=300s + + - name: Verify deployment + run: | + echo "=== Pods ===" + kubectl get pods -n rxcode-relay + echo "=== Service ===" + kubectl get service -n rxcode-relay + echo "=== Ingress ===" + kubectl get ingress -n rxcode-relay diff --git a/.gitignore b/.gitignore index 2c8829cb..42bd2b0a 100644 --- a/.gitignore +++ b/.gitignore @@ -38,3 +38,4 @@ relay-server/rxcode-relay relay-server/.env relay-server/.env.* relay-server/*.p8 +relay-server/k8s/secrets.yaml diff --git a/Packages/Sources/RxCodeChatKit/ChatMessageBubble.swift b/Packages/Sources/RxCodeChatKit/ChatMessageBubble.swift index 2e5ba255..24b6191d 100644 --- a/Packages/Sources/RxCodeChatKit/ChatMessageBubble.swift +++ b/Packages/Sources/RxCodeChatKit/ChatMessageBubble.swift @@ -82,7 +82,11 @@ private struct CompactChatMessageBubble: View { assistantText(text) } case .tool(let toolCall): - ToolResultView(toolCall: toolCall, isMessageStreaming: message.isStreaming) + if toolCall.name == "AskUserQuestion" { + AskUserQuestionView(toolCall: toolCall) + } else { + ToolResultView(toolCall: toolCall, isMessageStreaming: message.isStreaming) + } case .transientTools(_, let tools): ChatTransientToolSummaryView(tools: tools) } diff --git a/RxCode/App/AppState.swift b/RxCode/App/AppState.swift index 39952678..30b8007a 100644 --- a/RxCode/App/AppState.swift +++ b/RxCode/App/AppState.swift @@ -3601,7 +3601,16 @@ final class AppState { switch agentProvider { case .claudeCode: - mcpClaudeConfigPath = await mcp.writeClaudeConfig(projectPath: cwd) + // Allocate a per-session IDE-MCP port so the Claude agent can call + // IDE-only tools — cross-project chat (`ide__send_to_thread`), + // thread history, running jobs, usage. The bridge is a perl + // one-liner Claude runs as the `rxcode-ide` MCP server child. + let idePort = await ideMCPServer.allocate( + sessionKey: sessionKey, + capabilities: AgentProvider.claudeCode.staticCapabilities + ) + let bridge = idePort.map { IDEMCPServer.bridgeCommand(forPort: $0) } + mcpClaudeConfigPath = await mcp.writeClaudeConfig(projectPath: cwd, bridgeCommand: bridge) case .codex: mcpCodexOverrides = await mcp.codexConfigOverrides(projectPath: cwd) resolvedSendMode = registerMode diff --git a/RxCode/Services/ClaudeService.swift b/RxCode/Services/ClaudeService.swift index 36d48f2e..8fa5edb6 100644 --- a/RxCode/Services/ClaudeService.swift +++ b/RxCode/Services/ClaudeService.swift @@ -756,6 +756,39 @@ actor ClaudeCodeServer { // MARK: - Private Helpers + /// Extra system-prompt text appended via `--append-system-prompt`. Tells the + /// agent about the IDE-provided `rxcode-ide` MCP server, which lets it talk + /// to agents in other RxCode projects/threads and introspect editor state. + private static let ideToolsSystemPrompt = """ + # RxCode IDE tools + + You are running inside RxCode, a desktop IDE that hosts multiple projects, \ + each with its own chat threads and agents. RxCode injects a local MCP \ + server named `rxcode-ide`; its tools are exposed to you with the \ + `mcp__rxcode-ide__` prefix. Use them to coordinate with other agents \ + (multi-agent talk) and to read editor state: + + - `mcp__rxcode-ide__ide__get_projects` — list every project registered in \ + RxCode, so you can discover sibling projects to read or message. + - `mcp__rxcode-ide__ide__get_threads` — list or natural-language search \ + chat threads across projects (returns AI summaries, and ranked snippets \ + when a query is given). + - `mcp__rxcode-ide__ide__get_thread_messages` — fetch the message history \ + of a specific thread by id. + - `mcp__rxcode-ide__ide__send_to_thread` — talk to another project's agent: \ + send a prompt to an existing thread (`thread_id`) or start a new thread in \ + a project (`project_id`). This triggers a real agent run that may consume \ + tokens; it returns the other agent's reply. + - `mcp__rxcode-ide__ide__get_running_jobs` / `ide__get_job_output` — \ + inspect run-profile jobs (dev servers, scripts) executing in the IDE. + - `mcp__rxcode-ide__ide__get_usage` — current rate-limit / token usage. + + Reach for these when a task spans projects — e.g. checking what another \ + project already did, or delegating a subtask to its agent — rather than \ + guessing. Prefer reading threads first; only use `ide__send_to_thread` when \ + you actually need another agent to act, since it costs tokens. + """ + /// Build arguments array for the CLI invocation. /// /// The user prompt is NOT a CLI argument — it is written to stdin as a JSON @@ -801,6 +834,9 @@ actor ClaudeCodeServer { if let mcpConfigPath { args += ["--strict-mcp-config", "--mcp-config", mcpConfigPath] + // The `rxcode-ide` MCP server is part of this config — tell the + // agent the IDE multi-agent / introspection tools exist. + args += ["--append-system-prompt", Self.ideToolsSystemPrompt] } if let sessionId { diff --git a/RxCode/Services/MCPService.swift b/RxCode/Services/MCPService.swift index 9f2e9eb0..9197101b 100644 --- a/RxCode/Services/MCPService.swift +++ b/RxCode/Services/MCPService.swift @@ -130,11 +130,26 @@ actor MCPService { // MARK: - Provider Config - func writeClaudeConfig(projectPath: String?) async -> String? { + /// Write the `--mcp-config` JSON handed to the Claude CLI. If + /// `bridgeCommand` is non-nil an extra `rxcode-ide` stdio server is added — + /// the local MCP polyfill / introspection server that exposes IDE-only + /// tools (cross-project chat, thread history, running jobs, usage). + func writeClaudeConfig( + projectPath: String?, + bridgeCommand: (command: String, args: [String])? = nil + ) async -> String? { do { let records = try enabledRecords(projectPath: projectPath) + var servers = dictionaryForClaude(records) + if let bridge = bridgeCommand { + servers["rxcode-ide"] = [ + "type": "stdio", + "command": bridge.command, + "args": bridge.args, + ] + } let data = try JSONSerialization.data( - withJSONObject: ["mcpServers": dictionaryForClaude(records)], + withJSONObject: ["mcpServers": servers], options: [.prettyPrinted, .sortedKeys] ) return try writeGeneratedConfig(data: data, filename: "claude-mcp.json") diff --git a/RxCode/Services/RelayPresetCatalog.swift b/RxCode/Services/RelayPresetCatalog.swift new file mode 100644 index 00000000..e632acbd --- /dev/null +++ b/RxCode/Services/RelayPresetCatalog.swift @@ -0,0 +1,72 @@ +import Foundation +import Observation +import os.log + +/// One hosted relay-server option offered in the Mobile settings tab. +/// +/// The canonical list is published as `relays.json` on the RxCode website so +/// new deployments can be advertised without shipping an app update. +struct RelayPreset: Codable, Identifiable, Hashable, Sendable { + let id: String + let name: String + let url: String + var description: String? + var recommended: Bool? + + /// The parsed relay URL, or `nil` if the published value is malformed. + var relayURL: URL? { URL(string: url) } +} + +/// Loads the hosted relay catalog from `code.rxlab.app/relays.json`, falling +/// back to a bundled default so the picker still works offline. +@MainActor +@Observable +final class RelayPresetCatalog { + + static let shared = RelayPresetCatalog() + + private(set) var presets: [RelayPreset] = RelayPresetCatalog.bundledPresets + private(set) var isLoading = false + + private let logger = Logger(subsystem: "com.idealapp.RxCode", category: "RelayPresets") + private static let catalogURL = URL(string: "https://code.rxlab.app/relays.json")! + + /// Hardcoded fallback. Keep in sync with `website/public/relays.json`. + static let bundledPresets: [RelayPreset] = [ + RelayPreset( + id: "rxlab-hosted", + name: "RxLab Hosted Relay", + url: "wss://relay.code.rxlab.app", + description: "Official end-to-end encrypted relay operated by RxLab. Recommended for most setups.", + recommended: true + ) + ] + + /// Fetch the latest catalog. Keeps the current list on any failure. + func refresh() async { + guard !isLoading else { return } + isLoading = true + defer { isLoading = false } + do { + var request = URLRequest(url: Self.catalogURL) + request.cachePolicy = .reloadRevalidatingCacheData + request.timeoutInterval = 10 + let (data, response) = try await URLSession.shared.data(for: request) + guard let http = response as? HTTPURLResponse, (200..<300).contains(http.statusCode) else { + logger.warning("relay catalog fetch returned non-2xx response") + return + } + let catalog = try JSONDecoder().decode(RelayCatalog.self, from: data) + let valid = catalog.relays.filter { $0.relayURL != nil } + if !valid.isEmpty { + presets = valid + } + } catch { + logger.warning("relay catalog fetch failed: \(error.localizedDescription, privacy: .public)") + } + } +} + +private struct RelayCatalog: Codable { + let relays: [RelayPreset] +} diff --git a/RxCode/Views/Settings/MobileSettingsTab.swift b/RxCode/Views/Settings/MobileSettingsTab.swift index 0408b209..2e07cda3 100644 --- a/RxCode/Views/Settings/MobileSettingsTab.swift +++ b/RxCode/Views/Settings/MobileSettingsTab.swift @@ -3,15 +3,43 @@ import CoreImage.CIFilterBuiltins import RxCodeCore import RxCodeSync +/// How the relay server is chosen in the Mobile settings tab. +private enum RelayMode: Hashable { + /// One of the RxLab-hosted relays from the published catalog. + case hosted + /// A free-form relay URL typed by the user (e.g. a self-hosted relay). + case custom +} + /// "Mobile" tab in SettingsView. Lists paired iOS / iPadOS devices and lets /// the user pair a new one via QR code or unpair existing devices. struct MobileSettingsTab: View { @StateObject private var sync = MobileSyncService.shared + @State private var catalog = RelayPresetCatalog.shared @State private var showPairingSheet = false - @State private var relayURLText: String = MobileSyncService.shared.relayURL.absoluteString + @State private var relayMode: RelayMode + @State private var selectedPresetID: String? + @State private var customURLText: String @State private var testNotificationDeviceID: String? @State private var testNotificationAlert: TestNotificationAlert? + init() { + let current = MobileSyncService.shared.relayURL + let match = RelayPresetCatalog.bundledPresets.first { + MobileSettingsTab.normalize($0.url) == MobileSettingsTab.normalize(current.absoluteString) + } + _customURLText = State(initialValue: current.absoluteString) + if let match { + _relayMode = State(initialValue: .hosted) + _selectedPresetID = State(initialValue: match.id) + } else { + _relayMode = State(initialValue: .custom) + let fallback = RelayPresetCatalog.bundledPresets.first { $0.recommended == true } + ?? RelayPresetCatalog.bundledPresets.first + _selectedPresetID = State(initialValue: fallback?.id) + } + } + var body: some View { ScrollView { VStack(alignment: .leading, spacing: 20) { @@ -22,6 +50,10 @@ struct MobileSettingsTab: View { } .padding(20) } + .task { + await catalog.refresh() + reconcileWithCatalog() + } .sheet(isPresented: $showPairingSheet) { PairingSheet(sync: sync) } @@ -46,26 +78,143 @@ struct MobileSettingsTab: View { } private var relaySection: some View { - VStack(alignment: .leading, spacing: 8) { + VStack(alignment: .leading, spacing: 10) { Text("Relay server") .font(.headline) - Text("All sync traffic flows through this relay, end-to-end encrypted. Set up your own at github.com/rxlab/rxcode-relay for self-hosting.") + Text("All sync traffic flows through this relay, end-to-end encrypted. Use an RxLab-hosted relay, or point RxCode at your own — self-host with github.com/rxlab/rxcode-relay.") .font(.caption) .foregroundStyle(.secondary) - HStack { - TextField("ws://host:port", text: $relayURLText) - .textFieldStyle(.roundedBorder) - Button("Apply") { - if let url = URL(string: relayURLText) { - sync.updateRelay(url: url) - } + + Picker("Relay mode", selection: $relayMode) { + Text("Hosted server").tag(RelayMode.hosted) + Text("Custom URL").tag(RelayMode.custom) + } + .pickerStyle(.segmented) + .labelsHidden() + + switch relayMode { + case .hosted: + hostedRelayPicker + case .custom: + customRelayField + } + + HStack(spacing: 10) { + Button("Apply", action: applyRelay) + .disabled(!canApply) + if relayMode == .hosted, catalog.isLoading { + ProgressView().controlSize(.small) } - .disabled(URL(string: relayURLText) == nil) + Spacer() } + connectionBadge } } + @ViewBuilder + private var hostedRelayPicker: some View { + if catalog.presets.isEmpty { + Text("No hosted relays are available right now. Switch to Custom URL to enter one manually.") + .font(.caption) + .foregroundStyle(.secondary) + } else { + Picker("Hosted relay", selection: $selectedPresetID) { + ForEach(catalog.presets) { preset in + Text(preset.name).tag(Optional(preset.id)) + } + } + .labelsHidden() + + if let preset = selectedPreset { + VStack(alignment: .leading, spacing: 2) { + Text(preset.url) + .font(.caption) + .monospaced() + .foregroundStyle(.secondary) + .textSelection(.enabled) + if let description = preset.description { + Text(description) + .font(.caption) + .foregroundStyle(.secondary) + } + } + } + } + } + + private var customRelayField: some View { + VStack(alignment: .leading, spacing: 4) { + TextField("wss://host:port", text: $customURLText) + .textFieldStyle(.roundedBorder) + if !customURLText.trimmingCharacters(in: .whitespacesAndNewlines).isEmpty, + parsedCustomURL == nil { + Text("Enter a valid ws:// or wss:// URL.") + .font(.caption) + .foregroundStyle(.red) + } + } + } + + // MARK: - Relay selection helpers + + private var selectedPreset: RelayPreset? { + catalog.presets.first { $0.id == selectedPresetID } + } + + /// The custom URL parsed and validated as a relay endpoint, or `nil`. + private var parsedCustomURL: URL? { + let trimmed = customURLText.trimmingCharacters(in: .whitespacesAndNewlines) + guard let url = URL(string: trimmed), + let scheme = url.scheme?.lowercased(), + scheme == "ws" || scheme == "wss", + url.host?.isEmpty == false else { return nil } + return url + } + + /// The relay URL implied by the current mode and selection. + private var pendingRelayURL: URL? { + switch relayMode { + case .hosted: selectedPreset?.relayURL + case .custom: parsedCustomURL + } + } + + private var canApply: Bool { + guard let pending = pendingRelayURL else { return false } + return Self.normalize(pending.absoluteString) != Self.normalize(sync.relayURL.absoluteString) + } + + private func applyRelay() { + guard let url = pendingRelayURL else { return } + sync.updateRelay(url: url) + } + + /// After the live catalog loads, keep a valid preset selected and upgrade a + /// "custom" selection to the hosted picker when the active relay turns out + /// to be a freshly published preset. + private func reconcileWithCatalog() { + if selectedPreset == nil { + selectedPresetID = catalog.presets.first { $0.recommended == true }?.id + ?? catalog.presets.first?.id + } + guard relayMode == .custom, + customURLText == sync.relayURL.absoluteString, + let match = catalog.presets.first(where: { + Self.normalize($0.url) == Self.normalize(sync.relayURL.absoluteString) + }) + else { return } + relayMode = .hosted + selectedPresetID = match.id + } + + /// Normalize a URL string for equality checks (lowercased, no trailing `/`). + static func normalize(_ urlString: String) -> String { + var value = urlString.trimmingCharacters(in: .whitespacesAndNewlines).lowercased() + while value.hasSuffix("/") { value.removeLast() } + return value + } + @ViewBuilder private var connectionBadge: some View { switch sync.connectionState { diff --git a/relay-server/Dockerfile b/relay-server/Dockerfile index 9d0b8246..e82ccb4c 100644 --- a/relay-server/Dockerfile +++ b/relay-server/Dockerfile @@ -1,11 +1,12 @@ -FROM golang:1.22-alpine AS build +FROM golang:1.24-alpine AS build WORKDIR /src COPY go.mod go.sum* ./ RUN go mod download COPY . . RUN CGO_ENABLED=0 go build -o /out/relay . -FROM gcr.io/distroless/static +FROM gcr.io/distroless/static:nonroot COPY --from=build /out/relay /relay EXPOSE 8787 +USER nonroot:nonroot ENTRYPOINT ["/relay"] diff --git a/relay-server/README.md b/relay-server/README.md index de1a4311..9a45c792 100644 --- a/relay-server/README.md +++ b/relay-server/README.md @@ -54,6 +54,7 @@ missing file is non-fatal — the relay just uses whatever's in the process env. | `-apns-team-id` | `APNS_TEAM_ID` | 10-char Team ID. | | `-apns-topic` | `APNS_TOPIC` | iOS app bundle identifier (e.g. `app.rxlab.rxcodemobile`). | | `-apns-production` | `APNS_PRODUCTION` | `true` for production endpoint, else sandbox. | +| `-redis-url` | `REDIS_URL` | Redis URL for the multi-node backplane. Empty = single-node. | `APNS_KEY_B64` wins over `APNS_KEY_PATH` when both are set. Both standard and URL-safe base64 are accepted, and embedded whitespace/newlines are stripped — @@ -116,6 +117,37 @@ No bind-mount needed — the key lives only in the container environment. - The auth key file is sensitive — mount it via a secrets manager / Docker secret in production. Never commit `.p8` files. +## Scaling — single-node vs. multi-node + +The relay keeps live WebSocket connections in an in-memory map, so a naive +multi-replica deployment would drop messages: a desktop on pod A and its mobile +peer on pod B never share a connection map. + +- **Single-node (default).** Leave `REDIS_URL` unset. One Go process handles + thousands of WebSocket connections comfortably. Routing is purely in-memory; + a recipient that isn't connected gets a `delivery_failed` notice to the + sender. Run exactly **1 replica**. +- **Multi-node.** Set `REDIS_URL`. The relay uses Redis as a pub/sub backplane: + an envelope whose recipient isn't on the local pod is published to the + `relay:route` channel, and the pod holding that connection delivers it. + Per-pubkey presence keys (`relay:presence:*`, TTL-refreshed on every ping) + keep the `delivery_failed` notice accurate cluster-wide. Scale to any replica + count — no sticky sessions required. + +`/healthz` reports the active mode (`"mode": "single-node" | "multi-node"`). + +## Kubernetes + +Manifests for a multi-replica deployment live in [`k8s/`](./k8s/) — Deployment +(2 replicas), Service, Ingress (WebSocket-friendly timeouts), HPA, and a +PodDisruptionBudget. They wire `REDIS_URL` to the cluster-level Redis so the +backplane is on by default. See [`k8s/README.md`](./k8s/README.md). + +Creating a GitHub **release** builds, pushes, and deploys the relay via the +`Deploy Relay Server` GitHub Actions workflow +(`.github/workflows/relay-deploy.yaml`). Pushes to `main` and pull requests +touching `relay-server/**` build and smoke-test the image only. + ## Wire envelope ```json diff --git a/relay-server/backplane.go b/relay-server/backplane.go new file mode 100644 index 00000000..9ac9d1d7 --- /dev/null +++ b/relay-server/backplane.go @@ -0,0 +1,139 @@ +package main + +import ( + "context" + "crypto/rand" + "encoding/hex" + "encoding/json" + "log" + "time" + + "github.com/redis/go-redis/v9" +) + +const ( + // redisRouteChannel is the pub/sub channel every relay replica subscribes + // to. An envelope whose recipient is not connected to the publishing pod + // is fanned out here so whichever pod holds that connection delivers it. + redisRouteChannel = "relay:route" + + // redisPresencePrefix namespaces the per-pubkey presence keys that let any + // pod answer "is this peer connected anywhere in the cluster?" — needed to + // keep the delivery_failed notice accurate once routing spans pods. + redisPresencePrefix = "relay:presence:" + + // presenceTTL outlives pongWait so a key never expires under a healthy + // connection; writePump refreshes it every pingPeriod. + presenceTTL = pongWait + 60*time.Second +) + +// presenceRelease deletes a presence key only when it still belongs to the +// releasing pod. Without the compare, a peer that migrated to another replica +// (the new pod re-SET the key) would have its presence wiped by the old pod's +// disconnect cleanup, briefly looking offline cluster-wide. +var presenceRelease = redis.NewScript(` +if redis.call("get", KEYS[1]) == ARGV[1] then + return redis.call("del", KEYS[1]) +end +return 0 +`) + +// Backplane fans envelopes across relay replicas over Redis pub/sub and tracks +// cluster-wide connection presence. It is constructed only when REDIS_URL is +// set; otherwise the relay runs single-node and the hub keeps its purely +// in-memory behavior with no Redis dependency at runtime. +type Backplane struct { + rdb *redis.Client + hub *Hub + ctx context.Context + instanceID string +} + +// NewBackplane connects to Redis from a redis:// or rediss:// URL. A failed +// connection is fatal: if the operator asked for multi-node mode, silently +// degrading to single-node would drop every cross-pod message. +func NewBackplane(ctx context.Context, redisURL string, hub *Hub) (*Backplane, error) { + opt, err := redis.ParseURL(redisURL) + if err != nil { + return nil, err + } + rdb := redis.NewClient(opt) + if err := rdb.Ping(ctx).Err(); err != nil { + _ = rdb.Close() + return nil, err + } + return &Backplane{ + rdb: rdb, + hub: hub, + ctx: ctx, + instanceID: randomInstanceID(), + }, nil +} + +// Run subscribes to the route channel and hands every envelope to the local +// hub. An envelope for a peer connected to another replica finds no local +// connection and is dropped without re-publishing, so there is no fan-out loop +// — the pod that received it from a client is the only one that publishes. +func (b *Backplane) Run() { + sub := b.rdb.Subscribe(b.ctx, redisRouteChannel) + defer func() { _ = sub.Close() }() + log.Printf("backplane: subscribed to %s (instance %s)", redisRouteChannel, b.instanceID) + for msg := range sub.Channel() { + var env Envelope + if err := json.Unmarshal([]byte(msg.Payload), &env); err != nil { + log.Printf("backplane: malformed envelope: %v", err) + continue + } + b.hub.deliverLocal(&env) + } +} + +// Publish fans an envelope out to every other relay replica. +func (b *Backplane) Publish(env *Envelope) { + raw, err := json.Marshal(env) + if err != nil { + return + } + if err := b.rdb.Publish(b.ctx, redisRouteChannel, raw).Err(); err != nil { + log.Printf("backplane: publish to %s: %v", short(env.To), err) + } +} + +// MarkPresent records (or refreshes) a pubkey as connected to this pod. Called +// on register and on every ping so the key never expires while the socket is +// live. +func (b *Backplane) MarkPresent(pubkey string) { + if err := b.rdb.Set(b.ctx, redisPresencePrefix+pubkey, b.instanceID, presenceTTL).Err(); err != nil { + log.Printf("backplane: presence set %s: %v", short(pubkey), err) + } +} + +// ClearPresence removes a pubkey's presence marker on disconnect, but only if +// this pod still owns it (see presenceRelease). +func (b *Backplane) ClearPresence(pubkey string) { + if err := presenceRelease.Run(b.ctx, b.rdb, []string{redisPresencePrefix + pubkey}, b.instanceID).Err(); err != nil && err != redis.Nil { + log.Printf("backplane: presence clear %s: %v", short(pubkey), err) + } +} + +// ConnectedAnywhere reports whether a pubkey is connected to any relay +// replica. On a Redis error it returns true so the relay errs toward +// publishing rather than emitting a false delivery_failed. +func (b *Backplane) ConnectedAnywhere(pubkey string) bool { + n, err := b.rdb.Exists(b.ctx, redisPresencePrefix+pubkey).Result() + if err != nil { + log.Printf("backplane: presence check %s: %v", short(pubkey), err) + return true + } + return n > 0 +} + +// randomInstanceID returns a short per-process identifier used to scope +// presence ownership so one replica cannot clear another's keys. +func randomInstanceID() string { + var b [8]byte + if _, err := rand.Read(b[:]); err != nil { + return "relay" + } + return hex.EncodeToString(b[:]) +} diff --git a/relay-server/go.mod b/relay-server/go.mod index 61147108..5397475a 100644 --- a/relay-server/go.mod +++ b/relay-server/go.mod @@ -1,15 +1,18 @@ module github.com/rxlab/rxcode-relay -go 1.22 +go 1.24 require ( github.com/gorilla/websocket v1.5.3 github.com/joho/godotenv v1.5.1 + github.com/redis/go-redis/v9 v9.19.0 github.com/sideshow/apns2 v0.25.0 ) require ( + github.com/cespare/xxhash/v2 v2.3.0 // indirect github.com/golang-jwt/jwt/v4 v4.5.1 // indirect + go.uber.org/atomic v1.11.0 // indirect golang.org/x/net v0.30.0 // indirect golang.org/x/text v0.19.0 // indirect ) diff --git a/relay-server/go.sum b/relay-server/go.sum index ab5d0afa..a729e938 100644 --- a/relay-server/go.sum +++ b/relay-server/go.sum @@ -1,7 +1,14 @@ github.com/alecthomas/template v0.0.0-20190718012654-fb15b899a751/go.mod h1:LOuyumcjzFXgccqObfd/Ljyb9UuFJ6TxHnclSeseNhc= github.com/alecthomas/units v0.0.0-20201120081800-1786d5ef83d4/go.mod h1:OMCwj8VM1Kc9e19TLln2VL61YJF0x1XFtfdL4JdbSyE= -github.com/davecgh/go-spew v1.1.0 h1:ZDRjVQ15GmhC3fiQ8ni8+OwkZQO4DARzQgrnXU1Liz8= +github.com/bsm/ginkgo/v2 v2.12.0 h1:Ny8MWAHyOepLGlLKYmXG4IEkioBysk6GpaRTLC8zwWs= +github.com/bsm/ginkgo/v2 v2.12.0/go.mod h1:SwYbGRRDovPVboqFv0tPTcG1sN61LM1Z4ARdbAV9g4c= +github.com/bsm/gomega v1.27.10 h1:yeMWxP2pV2fG3FgAODIY8EiRE3dy0aeFYt4l7wh6yKA= +github.com/bsm/gomega v1.27.10/go.mod h1:JyEr/xRbxbtgWNi8tIEVPUYZ5Dzef52k01W3YH0H+O0= +github.com/cespare/xxhash/v2 v2.3.0 h1:UL815xU9SqsFlibzuggzjXhog7bL6oX9BbNZnL2UFvs= +github.com/cespare/xxhash/v2 v2.3.0/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs= github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= +github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c= +github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= github.com/golang-jwt/jwt/v4 v4.4.1/go.mod h1:m21LjoU+eqJr34lmDMbreY2eSTRJ1cv77w39/MY0Ch0= github.com/golang-jwt/jwt/v4 v4.5.1 h1:JdqV9zKUdtaa9gdPlywC3aeoEsR681PlKC+4F5gQgeo= github.com/golang-jwt/jwt/v4 v4.5.1/go.mod h1:m21LjoU+eqJr34lmDMbreY2eSTRJ1cv77w39/MY0Ch0= @@ -9,14 +16,22 @@ github.com/gorilla/websocket v1.5.3 h1:saDtZ6Pbx/0u+bgYQ3q96pZgCzfhKXGPqt7kZ72aN github.com/gorilla/websocket v1.5.3/go.mod h1:YR8l580nyteQvAITg2hZ9XVh4b55+EU/adAjf1fMHhE= github.com/joho/godotenv v1.5.1 h1:7eLL/+HRGLY0ldzfGMeQkb7vMd0as4CfYvUVzLqw0N0= github.com/joho/godotenv v1.5.1/go.mod h1:f4LDr5Voq0i2e/R5DDNOoa2zzDfwtkZa6DnEwAbqwq4= +github.com/klauspost/cpuid/v2 v2.2.10 h1:tBs3QSyvjDyFTq3uoc/9xFpCuOsJQFNPiAhYdw2skhE= +github.com/klauspost/cpuid/v2 v2.2.10/go.mod h1:hqwkgyIinND0mEev00jJYCxPNVRVXFQeu1XKlok6oO0= github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM= github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4= +github.com/redis/go-redis/v9 v9.19.0 h1:XPVaaPSnG6RhYf7p+rmSa9zZfeVAnWsH5h3lxthOm/k= +github.com/redis/go-redis/v9 v9.19.0/go.mod h1:v/M13XI1PVCDcm01VtPFOADfZtHf8YW3baQf57KlIkA= github.com/sideshow/apns2 v0.25.0 h1:XOzanncO9MQxkb03T/2uU2KcdVjYiIf0TMLzec0FTW4= github.com/sideshow/apns2 v0.25.0/go.mod h1:7Fceu+sL0XscxrfLSkAoH6UtvKefq3Kq1n4W3ayQZqE= github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME= github.com/stretchr/testify v1.4.0/go.mod h1:j7eGeouHqKxXV5pUuKE4zz7dFj8WfuZ+81PSLYec5m4= github.com/stretchr/testify v1.7.0 h1:nwc3DEeHmmLAfoZucVR881uASk0Mfjw8xYJ99tb5CcY= github.com/stretchr/testify v1.7.0/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg= +github.com/zeebo/xxh3 v1.1.0 h1:s7DLGDK45Dyfg7++yxI0khrfwq9661w9EN78eP/UZVs= +github.com/zeebo/xxh3 v1.1.0/go.mod h1:IisAie1LELR4xhVinxWS5+zf1lA4p0MW4T+w+W07F5s= +go.uber.org/atomic v1.11.0 h1:ZvwS0R+56ePWxUNi+Atn9dWONBPp/AUETXlHW0DxSjE= +go.uber.org/atomic v1.11.0/go.mod h1:LUxbIzbOniOlMKjJjyPfpl4v+PKK2cNJn91OQbhoJI0= golang.org/x/crypto v0.0.0-20170512130425-ab89591268e0/go.mod h1:6SG95UA2DQfeDnfUPMdvaQW0Q7yPrPDi9nlGo2tz2b4= golang.org/x/crypto v0.28.0 h1:GBDwsMXVQi34v5CCYUm2jkJvu4cbtru2U4TN2PSyQnw= golang.org/x/crypto v0.28.0/go.mod h1:rmgy+3RHxRZMyY0jjAJShp2zgEdOqj2AO7U0pYmeQ7U= @@ -25,6 +40,8 @@ golang.org/x/net v0.30.0 h1:AcW1SDZMkb8IpzCdQUaIq2sP4sZ4zw+55h6ynffypl4= golang.org/x/net v0.30.0/go.mod h1:2wGyMJ5iFasEhkwi13ChkO/t1ECNC4X4eBKkVFyYFlU= golang.org/x/sys v0.0.0-20210615035016-665e8c7367d1/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= golang.org/x/sys v0.0.0-20211216021012-1d35b9e2eb4e/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= +golang.org/x/sys v0.30.0 h1:QjkSwP/36a20jFYWkSue1YwXzLmsV5Gfq7Eiy72C1uc= +golang.org/x/sys v0.30.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA= golang.org/x/term v0.0.0-20210927222741-03fcf44c2211/go.mod h1:jbD1KX2456YbFQfuXm/mYQcufACuNUgVhRMnK/tPxf8= golang.org/x/text v0.3.7/go.mod h1:u+2+/6zg+i71rQMx5EYifcz6MCKuco9NR6JIITiCfzQ= golang.org/x/text v0.19.0 h1:kTxAhCbGbxhK0IwgSKiMO5awPoDQ0RpfiVYBfK860YM= diff --git a/relay-server/health.go b/relay-server/health.go index 62584e67..4cb80594 100644 --- a/relay-server/health.go +++ b/relay-server/health.go @@ -13,12 +13,17 @@ var startedAt = time.Now() // settings tab to verify the configured relay is reachable. func healthHandler(hub *Hub, sender *PushSender) http.HandlerFunc { return func(w http.ResponseWriter, r *http.Request) { + mode := "single-node" + if hub.backplane != nil { + mode = "multi-node" + } status := map[string]any{ "ok": true, "uptime_sec": int(time.Since(startedAt).Seconds()), "peers": hub.ConnectedCount(), "apns": sender != nil, - "version": "0.1.0", + "mode": mode, + "version": "0.2.0", } w.Header().Set("Content-Type", "application/json") w.WriteHeader(http.StatusOK) diff --git a/relay-server/k8s/README.md b/relay-server/k8s/README.md new file mode 100644 index 00000000..4e65cf26 --- /dev/null +++ b/relay-server/k8s/README.md @@ -0,0 +1,84 @@ +# Kubernetes Deployment — RxCode Relay + +Deploys the RxCode relay server (WebSocket sync relay + APNs forwarder) as a +horizontally scalable service. + +## Architecture + +- **rxcode-relay** — Go relay, 2+ replicas, port 8787 (`/ws`, `/push`, `/healthz`) +- **Redis backplane** — reuses the cluster-level Redis at + `redis://redis.redis.svc.cluster.local:6379` (namespace `redis`) for pub/sub + routing across replicas, so desktop and mobile peers may land on different + pods. Without `REDIS_URL` the relay runs single-node (1 replica only). + +Everything lives in the `rxcode-relay` namespace. + +## Files + +| File | Purpose | +|------|---------| +| `namespace.yaml` | `rxcode-relay` namespace | +| `configmap.yaml` | Non-secret config — `RELAY_ADDR`, `APNS_TOPIC`, `APNS_PRODUCTION`, `REDIS_URL` | +| `secrets.yaml` | APNs auth credentials (placeholder values — **not** in kustomization) | +| `deployment.yaml` | Relay Deployment, 2 replicas, `/healthz` probes | +| `service.yaml` | ClusterIP on port 8787 | +| `ingress.yaml` | NGINX ingress for `relay.rxlab.app` with WebSocket timeouts | +| `pdb.yaml` | PodDisruptionBudget — keeps ≥1 relay through drains/updates | +| `hpa.yaml` | HorizontalPodAutoscaler — CPU-based, 2–6 replicas (needs metrics-server) | +| `kustomization.yaml` | Kustomize orchestration | + +## Prerequisites + +1. Kubernetes cluster with `kubectl` access +2. NGINX Ingress Controller +3. cert-manager with a `letsencrypt-prod` cluster-issuer +4. Cluster-level Redis reachable at `redis.redis.svc.cluster.local:6379` +5. metrics-server (only required for the HPA) + +## Setup + +### 1. Populate secrets + +Edit `secrets.yaml` with real APNs values, then apply it manually (it is +deliberately excluded from `kustomization.yaml`): + +```bash +# encode the .p8 auth key +base64 < AuthKey_XXXXXXXXXX.p8 | tr -d '\n' + +kubectl apply -f secrets.yaml +``` + +### 2. Deploy + +```bash +kubectl apply -k . +``` + +### 3. Verify + +```bash +kubectl get pods -n rxcode-relay +kubectl get svc,ingress,hpa -n rxcode-relay +``` + +## CI/CD + +`.github/workflows/relay-deploy.yaml` deploys on every GitHub **release**: + +1. Builds the relay Docker image and pushes it to + `ghcr.io/rxtech-lab/rxcode-relay` (tags: release semver + `latest`). +2. Applies these manifests with `kubectl kustomize`, then pins the Deployment + to the released image tag and waits for the rollout. + +Pushes to `main` and pull requests touching `relay-server/**` build the image +and smoke-test `/healthz` only — no push, no deploy. + +Required GitHub secret: `K8S_CONFIG_FILE_B64` (base64-encoded kubeconfig). + +## Health & scaling + +- `GET /healthz` returns `{"mode": "multi-node", "peers": N, ...}`. +- Replica count is safe to raise: the Redis backplane routes envelopes across + pods, and no sticky sessions are needed. The HPA scales on CPU between 2 and + 6 replicas. diff --git a/relay-server/k8s/configmap.yaml b/relay-server/k8s/configmap.yaml new file mode 100644 index 00000000..7d59ab40 --- /dev/null +++ b/relay-server/k8s/configmap.yaml @@ -0,0 +1,17 @@ +apiVersion: v1 +kind: ConfigMap +metadata: + name: rxcode-relay-config + namespace: rxcode-relay + labels: + app: rxcode-relay +data: + # Listen address inside the pod. The Service targets this port. + RELAY_ADDR: ":8787" + # APNs topic must equal the RxCode Mobile bundle identifier exactly. + APNS_TOPIC: "com.idealapp.RxCode.Mobile" + APNS_PRODUCTION: "true" + # Cluster-level Redis (namespace `redis`), reused as the pub/sub backplane + # so the relay can run multiple replicas. Reachable from any namespace via + # this FQDN; auth-less. Remove this key to fall back to single-node mode. + REDIS_URL: "redis://redis.redis.svc.cluster.local:6379" diff --git a/relay-server/k8s/deployment.yaml b/relay-server/k8s/deployment.yaml new file mode 100644 index 00000000..cab9ede3 --- /dev/null +++ b/relay-server/k8s/deployment.yaml @@ -0,0 +1,68 @@ +apiVersion: apps/v1 +kind: Deployment +metadata: + name: rxcode-relay + namespace: rxcode-relay + labels: + app: rxcode-relay +spec: + # Safe to scale: the Redis backplane (REDIS_URL) routes envelopes across + # replicas, so a desktop and mobile peer may land on different pods. + replicas: 2 + strategy: + type: RollingUpdate + rollingUpdate: + maxUnavailable: 0 + maxSurge: 1 + selector: + matchLabels: + app: rxcode-relay + template: + metadata: + labels: + app: rxcode-relay + spec: + containers: + - name: relay + image: ghcr.io/rxtech-lab/rxcode-relay:latest + ports: + - containerPort: 8787 + protocol: TCP + envFrom: + - configMapRef: + name: rxcode-relay-config + - secretRef: + name: rxcode-relay-secrets + resources: + requests: + memory: "64Mi" + cpu: "50m" + limits: + memory: "256Mi" + cpu: "500m" + livenessProbe: + httpGet: + path: /healthz + port: 8787 + initialDelaySeconds: 10 + periodSeconds: 10 + timeoutSeconds: 5 + failureThreshold: 3 + readinessProbe: + httpGet: + path: /healthz + port: 8787 + initialDelaySeconds: 5 + periodSeconds: 5 + timeoutSeconds: 5 + failureThreshold: 3 + securityContext: + allowPrivilegeEscalation: false + runAsNonRoot: true + runAsUser: 65532 + runAsGroup: 65532 + readOnlyRootFilesystem: true + capabilities: + drop: + - ALL + restartPolicy: Always diff --git a/relay-server/k8s/hpa.yaml b/relay-server/k8s/hpa.yaml new file mode 100644 index 00000000..86e09bb0 --- /dev/null +++ b/relay-server/k8s/hpa.yaml @@ -0,0 +1,26 @@ +apiVersion: autoscaling/v2 +kind: HorizontalPodAutoscaler +metadata: + name: rxcode-relay + namespace: rxcode-relay + labels: + app: rxcode-relay +spec: + scaleTargetRef: + apiVersion: apps/v1 + kind: Deployment + name: rxcode-relay + minReplicas: 2 + maxReplicas: 6 + metrics: + - type: Resource + resource: + name: cpu + target: + type: Utilization + averageUtilization: 70 + behavior: + scaleDown: + # WebSocket clients reconnect onto surviving pods; let connections drain + # before removing capacity rather than thrashing replicas. + stabilizationWindowSeconds: 300 diff --git a/relay-server/k8s/ingress.yaml b/relay-server/k8s/ingress.yaml new file mode 100644 index 00000000..2a643d16 --- /dev/null +++ b/relay-server/k8s/ingress.yaml @@ -0,0 +1,34 @@ +apiVersion: networking.k8s.io/v1 +kind: Ingress +metadata: + name: rxcode-relay + namespace: rxcode-relay + labels: + app: rxcode-relay + annotations: + nginx.ingress.kubernetes.io/ssl-redirect: "true" + nginx.ingress.kubernetes.io/force-ssl-redirect: "true" + cert-manager.io/cluster-issuer: "letsencrypt-prod" + # WebSocket connections are long-lived: the relay only pings every 25s, so + # the proxy must not time out an otherwise idle /ws stream, and responses + # must stream through unbuffered. + nginx.ingress.kubernetes.io/proxy-read-timeout: "3600" + nginx.ingress.kubernetes.io/proxy-send-timeout: "3600" + nginx.ingress.kubernetes.io/proxy-buffering: "off" +spec: + ingressClassName: nginx + tls: + - hosts: + - relay.rxlab.app + secretName: rxcode-relay-tls + rules: + - host: relay.rxlab.app + http: + paths: + - path: / + pathType: Prefix + backend: + service: + name: rxcode-relay + port: + number: 8787 diff --git a/relay-server/k8s/kustomization.yaml b/relay-server/k8s/kustomization.yaml new file mode 100644 index 00000000..c621fefd --- /dev/null +++ b/relay-server/k8s/kustomization.yaml @@ -0,0 +1,23 @@ +apiVersion: kustomize.config.k8s.io/v1beta1 +kind: Kustomization + +metadata: + name: rxcode-relay + namespace: rxcode-relay + +resources: + - namespace.yaml + # secrets.yaml is intentionally excluded — apply it manually after filling in + # real APNs values: `kubectl apply -f k8s/secrets.yaml`. + - configmap.yaml + - deployment.yaml + - service.yaml + - ingress.yaml + - pdb.yaml + - hpa.yaml + +namespace: rxcode-relay + +images: + - name: ghcr.io/rxtech-lab/rxcode-relay + newTag: latest diff --git a/relay-server/k8s/namespace.yaml b/relay-server/k8s/namespace.yaml new file mode 100644 index 00000000..30b89351 --- /dev/null +++ b/relay-server/k8s/namespace.yaml @@ -0,0 +1,8 @@ +apiVersion: v1 +kind: Namespace +metadata: + name: rxcode-relay + labels: + name: rxcode-relay + app.kubernetes.io/name: rxcode-relay + app.kubernetes.io/managed-by: kustomize diff --git a/relay-server/k8s/pdb.yaml b/relay-server/k8s/pdb.yaml new file mode 100644 index 00000000..12553fcc --- /dev/null +++ b/relay-server/k8s/pdb.yaml @@ -0,0 +1,14 @@ +apiVersion: policy/v1 +kind: PodDisruptionBudget +metadata: + name: rxcode-relay + namespace: rxcode-relay + labels: + app: rxcode-relay +spec: + # Keep at least one relay serving WebSocket clients through node drains and + # rolling updates. + minAvailable: 1 + selector: + matchLabels: + app: rxcode-relay diff --git a/relay-server/k8s/service.yaml b/relay-server/k8s/service.yaml new file mode 100644 index 00000000..daf8f4de --- /dev/null +++ b/relay-server/k8s/service.yaml @@ -0,0 +1,16 @@ +apiVersion: v1 +kind: Service +metadata: + name: rxcode-relay + namespace: rxcode-relay + labels: + app: rxcode-relay +spec: + type: ClusterIP + ports: + - name: http + port: 8787 + targetPort: 8787 + protocol: TCP + selector: + app: rxcode-relay diff --git a/relay-server/main.go b/relay-server/main.go index d068ad8f..de6d6847 100644 --- a/relay-server/main.go +++ b/relay-server/main.go @@ -39,6 +39,7 @@ func main() { apnsTeamID := flag.String("apns-team-id", os.Getenv("APNS_TEAM_ID"), "APNs Team ID (env: APNS_TEAM_ID)") apnsTopic := flag.String("apns-topic", os.Getenv("APNS_TOPIC"), "APNs topic / iOS bundle ID (env: APNS_TOPIC)") apnsProduction := flag.Bool("apns-production", envBool("APNS_PRODUCTION", false), "use production APNs endpoint instead of sandbox (env: APNS_PRODUCTION)") + redisURL := flag.String("redis-url", os.Getenv("REDIS_URL"), "Redis URL for the multi-node pub/sub backplane; empty runs single-node (env: REDIS_URL)") flag.Parse() // APNs .p8 key can come from a base64-encoded env var (preferred for @@ -49,6 +50,22 @@ func main() { } hub := NewHub() + + // Optional Redis backplane. With REDIS_URL set the relay can run multiple + // replicas behind a load balancer; without it the relay runs single-node + // with purely in-memory routing and no Redis dependency at runtime. + if *redisURL != "" { + bp, err := NewBackplane(context.Background(), *redisURL, hub) + if err != nil { + log.Fatalf("init redis backplane: %v", err) + } + hub.backplane = bp + go bp.Run() + log.Printf("redis backplane enabled — multi-node mode") + } else { + log.Printf("no REDIS_URL set — single-node mode (in-memory routing only)") + } + go hub.Run() var pushSender *PushSender diff --git a/relay-server/relay.go b/relay-server/relay.go index 9b915bd7..96c0d113 100644 --- a/relay-server/relay.go +++ b/relay-server/relay.go @@ -49,6 +49,10 @@ type Hub struct { register chan *Conn unregister chan *Conn route chan *Envelope + + // backplane fans envelopes to other replicas over Redis. nil means the + // relay runs single-node and routes purely from the in-memory map. + backplane *Backplane } func NewHub() *Hub { @@ -70,14 +74,25 @@ func (h *Hub) Run() { } h.conns[c.pubkey] = c h.mu.Unlock() + if h.backplane != nil { + h.backplane.MarkPresent(c.pubkey) + } log.Printf("registered %s", short(c.pubkey)) case c := <-h.unregister: h.mu.Lock() + removed := false if current, ok := h.conns[c.pubkey]; ok && current == c { delete(h.conns, c.pubkey) close(c.send) + removed = true } h.mu.Unlock() + // Only clear cluster presence when this exact connection was the + // one removed — a reconnect that already replaced it must keep its + // presence marker intact. + if removed && h.backplane != nil { + h.backplane.ClearPresence(c.pubkey) + } log.Printf("unregistered %s", short(c.pubkey)) case env := <-h.route: h.deliver(env) @@ -85,19 +100,38 @@ func (h *Hub) Run() { } } +// deliver routes an envelope to its recipient. The recipient may be connected +// to this pod, to another pod (reached via the Redis backplane), or offline. func (h *Hub) deliver(env *Envelope) { + if h.deliverLocal(env) { + return + } + // Not connected to this pod. + if h.backplane != nil { + // Another replica may hold the connection. The backplane round-trips + // to Redis, so run it off the hub loop to keep local routing fast. + go h.routeViaBackplane(env) + return + } + // Single-node: a missing local connection means the peer is offline. + h.notifyDeliveryFailed(env.From, env.To) +} + +// deliverLocal delivers an envelope to a recipient connected to this pod and +// reports whether such a connection existed. It never touches the backplane, +// so it is safe to call on envelopes arriving from the backplane without +// risking a re-publish loop. +func (h *Hub) deliverLocal(env *Envelope) bool { raw, err := json.Marshal(env) if err != nil { log.Printf("marshal envelope: %v", err) - return + return false } h.mu.RLock() dest, ok := h.conns[env.To] h.mu.RUnlock() if !ok { - // Drop on offline. Optionally notify sender. - h.notifyDeliveryFailed(env.From, env.To) - return + return false } select { case dest.send <- raw: @@ -105,6 +139,16 @@ func (h *Hub) deliver(env *Envelope) { // Slow consumer; drop and disconnect. log.Printf("send buffer full for %s — dropping", short(env.To)) } + return true +} + +// routeViaBackplane fans an envelope to the other replicas and, when the peer +// is offline cluster-wide, notifies the sender. Runs on its own goroutine. +func (h *Hub) routeViaBackplane(env *Envelope) { + h.backplane.Publish(env) + if !h.backplane.ConnectedAnywhere(env.To) { + h.notifyDeliveryFailed(env.From, env.To) + } } func (h *Hub) notifyDeliveryFailed(from, to string) { @@ -209,6 +253,10 @@ func (c *Conn) writePump() { return } case <-ticker.C: + // Refresh cluster presence before it can expire (TTL > pingPeriod). + if c.hub.backplane != nil { + c.hub.backplane.MarkPresent(c.pubkey) + } _ = c.ws.SetWriteDeadline(time.Now().Add(writeWait)) if err := c.ws.WriteMessage(websocket.PingMessage, nil); err != nil { return diff --git a/website/.gitignore b/website/.gitignore index de08fad0..14e3ce86 100644 --- a/website/.gitignore +++ b/website/.gitignore @@ -44,3 +44,4 @@ next-env.d.ts package-lock.json yarn.lock pnpm-lock.yaml +.env*.local diff --git a/website/public/relays.json b/website/public/relays.json new file mode 100644 index 00000000..4d714c53 --- /dev/null +++ b/website/public/relays.json @@ -0,0 +1,13 @@ +{ + "version": 1, + "updatedAt": "2026-05-20", + "relays": [ + { + "id": "rxlab-hosted", + "name": "RxLab Hosted Relay", + "url": "wss://relay.code.rxlab.app", + "description": "Official end-to-end encrypted relay operated by RxLab. Recommended for most setups.", + "recommended": true + } + ] +}