feat(sync): add Helm chart and WebSocket push for SuperSync (#6971)

* feat(sync): add Helm chart for SuperSync Kubernetes deployment

Includes Deployment, StatefulSet (PostgreSQL), Service, Ingress,
ConfigMap, Secret, HPA, PDB, NetworkPolicy, and test templates.
Supports both bundled PostgreSQL and external database configurations.

* feat(sync): add WebSocket push notifications for near-realtime sync

Server: Fastify WebSocket plugin with connection manager, app-level
heartbeat (30s), debounced notifications, and per-user routing.
Client: WebSocket service with exponential backoff reconnection,
WS-triggered download service, and reduced polling when connected.

* fix(sync): improve WebSocket error handling and reactivity

Make syncInterval$ reactive to WS connection state, fix Set/Map
mutation during heartbeat iteration, add error handling to WS route
handler, separate JSON parsing from message handling, detect auth
failures in WS-triggered downloads, and add logging to all empty
catch blocks.

* fix(sync): address PR review findings for WebSocket and Helm

Fix race condition in WS-triggered download pipeline by moving
isSyncInProgress filter after debounce and adding guard in
_downloadOps. Add logging to remaining empty catch blocks, fix
missing $NODE_IP in Helm NOTES.txt, and correct inaccurate
comments in values.yaml and ws-triggered-download.service.ts.

* fix(sync): add rate limiting to WebSocket upgrade endpoint

Limit WS connection attempts to 10 per minute per IP to prevent
connection flooding, matching the rate-limit pattern used by other
sync endpoints.

* fix(sync): extract shared constants, fix debounce correctness, replace deprecated toPromise

Extract CLIENT_ID_REGEX and MAX_CLIENT_ID_LENGTH into sync.const.ts
to prevent drift between sync.routes.ts and websocket.routes.ts.

Fix debounce in notifyNewOps to accumulate excluded client IDs across
rapid calls from different clients, preventing self-notifications.

Replace deprecated toPromise() with firstValueFrom in connectWebSocket
and _sync methods. Add .catch() to reconnect path and logging to
silent early returns in _downloadOps.

* fix(build): restore correct package-lock.json and fix upstream lint errors

* revert: restore upstream HTML formatting to match CI prettier config

* fix(sync): use DownloadOutcome discriminated union in WsTriggeredDownloadService

* fix(sync): narrow TokenVerificationResult before accessing userId

* fix(sync): await async getProviderById in connectWebSocket

Missing await caused the Promise object to be cast to
SuperSyncProvider, making getWebSocketParams undefined.

* feat(sync): add SEED_USERS env var to create verified users on startup

For self-hosted single-user setups: set SEED_USERS=email1,email2 to
create verified users on boot and log their access tokens. Skips
existing users. Removes need for SMTP/magic link registration.

Also fix Dockerfile to use npm install instead of npm ci for lockfile
compatibility.

* fix(sync): address PR review feedback for Helm chart and server

Security:
- Remove seed-users.ts (logged full JWT tokens to stdout)
- Add fail assertions for missing jwtSecret and postgresql.password
- Add smtp.user/smtp.password values fields
- Add from: selector to NetworkPolicy ingress
- Add egress rule for external database when postgresql.enabled=false

Correctness:
- Fix PostgreSQL StatefulSet indentation for non-persistent mode
- Fix NOTES.txt panic on empty tls list (use len instead of index)
- Restore npm ci in Dockerfile by using node:24-alpine (ships npm 11)
- Add Recreate deployment strategy when using RWO PVC

Operational:
- Add fail guard preventing HPA maxReplicas > 1 (in-memory WS state)
- Fix PDB to use maxUnavailable instead of minAvailable
- Add WebSocket ingress timeout annotation examples
- Add Prisma db push init container for schema migrations
- Default serviceAccount.automount to false
- Add Chart.yaml maintainers, home, sources metadata

* feat(sync): add ALLOWED_EMAILS env var to restrict registration

Supports fully qualified emails (user@example.com) and domain
wildcards (*@example.com). When unset, all emails are allowed.
Applied to all three registration endpoints (passkey options,
passkey verify, magic link).

* fix(sync): exempt health endpoint from rate limiting

Kubernetes liveness/readiness probes hit /health every 5-15s,
exhausting the global rate limit (100 req/15min) and causing
429 responses that trigger container restarts.

* fix(sync): harden WebSocket, Helm chart, and sync services from multi-agent review

- Pin prisma@5.22.0 in init container and use migrate deploy instead of db push
- Add per-user WebSocket connection limit (max 10) to prevent resource exhaustion
- Add replicaCount > 1 fail guard in deployment template (in-memory WS state)
- Set maxPayload: 1024 on WebSocket plugin (only pong messages expected)
- Remove premature IN_SYNC status from download-only WS-triggered sync
- Fix double removeConnection on error+close events (let close handle cleanup)
- DRY debounce logic in notifyNewOps and store latestSeq on pending entry
- Remove dead fromClientId from NewOpsNotification and WsMessage interfaces
- Add takeUntilDestroyed to _wsProviderCleanup subscription
- Simplify email-allowlist.ts to eager init (eliminate mutable state)
- Guard connectWebSocket at call site for non-SuperSync providers
- Add distinctUntilChanged to syncInterval$ to prevent unnecessary resets
- Add container-level securityContext to PostgreSQL StatefulSet
- Fix HPA maxReplicas default to 1 (matches single-replica constraint)

* fix(sync): restore migration in Dockerfile CMD for non-Helm Docker users

Helm users get migration via init container (runs first, CMD becomes no-op).
Docker-compose/plain Docker users get automatic migration back on startup.

* fix(boards): add missing drag delay for touch and extract shared signal

Add cdkDragStartDelay to board-panel drag items, which was missing
entirely. Extract the repeated `isTouchActive() ? DRAG_DELAY_FOR_TOUCH : 0`
expression into a shared `dragDelayForTouch` computed signal and refactor
all 8 components to use it.

* test(sync): add comprehensive WebSocket test coverage

Add unit tests for the new WebSocket real-time sync notification pipeline:

- WebSocketConnectionService (server): connection lifecycle, max-per-user
  limits, notification debouncing, heartbeat, graceful shutdown (13 tests)
- WebSocket routes validation: token/clientId validation, close codes,
  error handling, validation ordering (18 tests)
- SuperSyncWebSocketService (frontend): reconnection with exponential
  backoff, heartbeat, disconnect cleanup, URL conversion (6 tests)
- WsTriggeredDownloadService: auth error handling, pipeline resilience,
  start() idempotency (3 tests)
- SyncWrapperService: connectWebSocket guards for non-SuperSync, null
  params, and already-connected cases (3 tests)
- E2E realtime push: verifies WS-triggered download between two clients

Quality fixes:
- Replace waitForTimeout with syncAndWait in E2E
- Add explanatory comment for microtask flushing in sync-wrapper spec
- Use getter for isSyncInProgress mock in ws-triggered-download spec

---------

Co-authored-by: Johannes Millan <johannes.millan@gmail.com>
This commit is contained in:
Thorsten Klein 2026-03-30 21:34:30 +02:00 committed by GitHub
parent 34b086b4c7
commit 7fa8f12132
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
42 changed files with 3079 additions and 29 deletions

View file

@ -6,3 +6,4 @@ node_modules
*.env.example *.env.example
.prettierignore .prettierignore
packages/super-sync-server/public/simplewebauthn-browser.min.js packages/super-sync-server/public/simplewebauthn-browser.min.js
packages/super-sync-server/helm

View file

@ -0,0 +1,72 @@
import { test, expect } from '../../fixtures/supersync.fixture';
import {
closeClient,
createSimulatedClient,
createTestUser,
getSuperSyncConfig,
waitForTask,
type SimulatedE2EClient,
} from '../../utils/supersync-helpers';
/**
* SuperSync realtime push E2E tests.
*
* Verifies the new WebSocket notification flow end-to-end:
* 1. Both clients complete an initial successful SuperSync sync
* 2. Client A uploads a change
* 3. Client B receives the change without clicking sync again
*
* This specifically guards the PR's new "upload -> WS notify -> download" path.
*/
test.describe('@supersync Realtime Push', () => {
test('propagates changes to another client without manual sync', async ({
browser,
baseURL,
testRunId,
}) => {
test.setTimeout(180000);
let clientA: SimulatedE2EClient | null = null;
let clientB: SimulatedE2EClient | null = null;
try {
const appUrl = baseURL || 'http://localhost:4242';
const user = await createTestUser(testRunId);
const syncConfig = getSuperSyncConfig(user);
clientA = await createSimulatedClient(browser, appUrl, 'A', testRunId);
await clientA.sync.setupSuperSync(syncConfig);
const baselineTask = `Realtime-Baseline-${testRunId}`;
await clientA.workView.addTask(baselineTask);
await clientA.sync.syncAndWait();
await waitForTask(clientA.page, baselineTask);
clientB = await createSimulatedClient(browser, appUrl, 'B', testRunId);
await clientB.sync.setupSuperSync(syncConfig);
await clientB.sync.syncAndWait();
await waitForTask(clientB.page, baselineTask);
// Allow the post-sync WebSocket connection to establish on both clients.
// A second syncAndWait round-trip ensures the WS-triggered download pipeline is active.
await clientA.sync.syncAndWait();
await clientB.sync.syncAndWait();
const pushedTask = `Realtime-Pushed-${testRunId}`;
await clientA.workView.addTask(pushedTask);
const pushStart = Date.now();
await clientA.sync.syncAndWait();
// No manual sync on client B here: it should update via WS-triggered download.
await waitForTask(clientB.page, pushedTask, 10000);
const propagationMs = Date.now() - pushStart;
expect(propagationMs).toBeLessThan(10000);
expect(await clientB.sync.hasSyncError()).toBe(false);
} finally {
if (clientA) await closeClient(clientA);
if (clientB) await closeClient(clientB);
}
});
});

48
package-lock.json generated
View file

@ -4939,6 +4939,27 @@
"url": "https://github.com/sponsors/isaacs" "url": "https://github.com/sponsors/isaacs"
} }
}, },
"node_modules/@fastify/websocket": {
"version": "11.2.0",
"resolved": "https://registry.npmjs.org/@fastify/websocket/-/websocket-11.2.0.tgz",
"integrity": "sha512-3HrDPbAG1CzUCqnslgJxppvzaAZffieOVbLp1DAy1huCSynUWPifSvfdEDUR8HlJLp3sp1A36uOM2tJogADS8w==",
"funding": [
{
"type": "github",
"url": "https://github.com/sponsors/fastify"
},
{
"type": "opencollective",
"url": "https://opencollective.com/fastify"
}
],
"license": "MIT",
"dependencies": {
"duplexify": "^4.1.3",
"fastify-plugin": "^5.0.0",
"ws": "^8.16.0"
}
},
"node_modules/@floating-ui/core": { "node_modules/@floating-ui/core": {
"version": "1.7.5", "version": "1.7.5",
"resolved": "https://registry.npmjs.org/@floating-ui/core/-/core-1.7.5.tgz", "resolved": "https://registry.npmjs.org/@floating-ui/core/-/core-1.7.5.tgz",
@ -14837,6 +14858,18 @@
"dev": true, "dev": true,
"license": "MIT" "license": "MIT"
}, },
"node_modules/duplexify": {
"version": "4.1.3",
"resolved": "https://registry.npmjs.org/duplexify/-/duplexify-4.1.3.tgz",
"integrity": "sha512-M3BmBhwJRZsSx38lZyhE53Csddgzl5R7xGJNk7CVddZD6CcmwMCH8J+7AprIrQKH7TonKxaCjcv27Qmf+sQ+oA==",
"license": "MIT",
"dependencies": {
"end-of-stream": "^1.4.1",
"inherits": "^2.0.3",
"readable-stream": "^3.1.1",
"stream-shift": "^1.0.2"
}
},
"node_modules/eastasianwidth": { "node_modules/eastasianwidth": {
"version": "0.2.0", "version": "0.2.0",
"resolved": "https://registry.npmjs.org/eastasianwidth/-/eastasianwidth-0.2.0.tgz", "resolved": "https://registry.npmjs.org/eastasianwidth/-/eastasianwidth-0.2.0.tgz",
@ -15287,7 +15320,6 @@
"version": "1.4.5", "version": "1.4.5",
"resolved": "https://registry.npmjs.org/end-of-stream/-/end-of-stream-1.4.5.tgz", "resolved": "https://registry.npmjs.org/end-of-stream/-/end-of-stream-1.4.5.tgz",
"integrity": "sha512-ooEGc6HP26xXq/N+GCGOT0JKCLDGrq2bQUZrQ7gyrJiZANJ/8YDTxTpQBXGMn+WbIQXNVpyWymm7KYVICQnyOg==", "integrity": "sha512-ooEGc6HP26xXq/N+GCGOT0JKCLDGrq2bQUZrQ7gyrJiZANJ/8YDTxTpQBXGMn+WbIQXNVpyWymm7KYVICQnyOg==",
"dev": true,
"license": "MIT", "license": "MIT",
"dependencies": { "dependencies": {
"once": "^1.4.0" "once": "^1.4.0"
@ -22778,7 +22810,6 @@
"version": "1.4.0", "version": "1.4.0",
"resolved": "https://registry.npmjs.org/once/-/once-1.4.0.tgz", "resolved": "https://registry.npmjs.org/once/-/once-1.4.0.tgz",
"integrity": "sha512-lNaJgI+2Q5URQBkccEKHTQOPaXdUxnZZElQTZY0MFUAuaEqe1E+Nyvgdz/aIyNi6Z9MzO5dv1H8n58/GELp3+w==", "integrity": "sha512-lNaJgI+2Q5URQBkccEKHTQOPaXdUxnZZElQTZY0MFUAuaEqe1E+Nyvgdz/aIyNi6Z9MzO5dv1H8n58/GELp3+w==",
"dev": true,
"license": "ISC", "license": "ISC",
"dependencies": { "dependencies": {
"wrappy": "1" "wrappy": "1"
@ -24482,7 +24513,6 @@
"version": "3.6.2", "version": "3.6.2",
"resolved": "https://registry.npmjs.org/readable-stream/-/readable-stream-3.6.2.tgz", "resolved": "https://registry.npmjs.org/readable-stream/-/readable-stream-3.6.2.tgz",
"integrity": "sha512-9u/sniCrY3D5WdsERHzHE4G2YCXqoG5FTHUiCC4SIbr6XcLZBY05ya9EKjYek9O5xOAwjGq+1JdGBAS7Q9ScoA==", "integrity": "sha512-9u/sniCrY3D5WdsERHzHE4G2YCXqoG5FTHUiCC4SIbr6XcLZBY05ya9EKjYek9O5xOAwjGq+1JdGBAS7Q9ScoA==",
"dev": true,
"license": "MIT", "license": "MIT",
"dependencies": { "dependencies": {
"inherits": "^2.0.3", "inherits": "^2.0.3",
@ -26438,6 +26468,12 @@
"duplexer": "~0.1.1" "duplexer": "~0.1.1"
} }
}, },
"node_modules/stream-shift": {
"version": "1.0.3",
"resolved": "https://registry.npmjs.org/stream-shift/-/stream-shift-1.0.3.tgz",
"integrity": "sha512-76ORR0DO1o1hlKwTbi/DM3EXWGf3ZJYO8cXX5RJwnul2DEg2oyoZyjLNoQM8WsvZiFKCRfC1O0J7iCvie3RZmQ==",
"license": "MIT"
},
"node_modules/streamroller": { "node_modules/streamroller": {
"version": "3.1.5", "version": "3.1.5",
"resolved": "https://registry.npmjs.org/streamroller/-/streamroller-3.1.5.tgz", "resolved": "https://registry.npmjs.org/streamroller/-/streamroller-3.1.5.tgz",
@ -26492,7 +26528,6 @@
"version": "1.3.0", "version": "1.3.0",
"resolved": "https://registry.npmjs.org/string_decoder/-/string_decoder-1.3.0.tgz", "resolved": "https://registry.npmjs.org/string_decoder/-/string_decoder-1.3.0.tgz",
"integrity": "sha512-hkRX8U1WjJFd8LsDJ2yQ/wWWxaopEsABU1XfkM8A+j0+85JAGppt16cr1Whg6KIbb4okU6Mql6BOj+uup/wKeA==", "integrity": "sha512-hkRX8U1WjJFd8LsDJ2yQ/wWWxaopEsABU1XfkM8A+j0+85JAGppt16cr1Whg6KIbb4okU6Mql6BOj+uup/wKeA==",
"dev": true,
"license": "MIT", "license": "MIT",
"dependencies": { "dependencies": {
"safe-buffer": "~5.2.0" "safe-buffer": "~5.2.0"
@ -28460,7 +28495,6 @@
"version": "1.0.2", "version": "1.0.2",
"resolved": "https://registry.npmjs.org/util-deprecate/-/util-deprecate-1.0.2.tgz", "resolved": "https://registry.npmjs.org/util-deprecate/-/util-deprecate-1.0.2.tgz",
"integrity": "sha512-EPD5q1uXyFxJpCrLnCc1nHnq3gOa6DZBocAIiI2TaSCA7VCJ1UJDMagCzIkXNsUYfD1daK//LTEQ8xiIbrHtcw==", "integrity": "sha512-EPD5q1uXyFxJpCrLnCc1nHnq3gOa6DZBocAIiI2TaSCA7VCJ1UJDMagCzIkXNsUYfD1daK//LTEQ8xiIbrHtcw==",
"dev": true,
"license": "MIT" "license": "MIT"
}, },
"node_modules/utils-merge": { "node_modules/utils-merge": {
@ -29804,7 +29838,6 @@
"version": "1.0.2", "version": "1.0.2",
"resolved": "https://registry.npmjs.org/wrappy/-/wrappy-1.0.2.tgz", "resolved": "https://registry.npmjs.org/wrappy/-/wrappy-1.0.2.tgz",
"integrity": "sha512-l4Sp/DRseor9wL6EvV2+TuQn63dMkPjZ/sp9XkghTEbV9KlPS1xUsZ3u7/IQO4wxtcFB4bgpQPRcR3QCvezPcQ==", "integrity": "sha512-l4Sp/DRseor9wL6EvV2+TuQn63dMkPjZ/sp9XkghTEbV9KlPS1xUsZ3u7/IQO4wxtcFB4bgpQPRcR3QCvezPcQ==",
"dev": true,
"license": "ISC" "license": "ISC"
}, },
"node_modules/write-file-atomic": { "node_modules/write-file-atomic": {
@ -29825,7 +29858,6 @@
"version": "8.18.3", "version": "8.18.3",
"resolved": "https://registry.npmjs.org/ws/-/ws-8.18.3.tgz", "resolved": "https://registry.npmjs.org/ws/-/ws-8.18.3.tgz",
"integrity": "sha512-PEIGCY5tSlUt50cqyMXfCzX+oOPqN0vuGqWzbcJ2xvnkzkq46oOpz7dQaTDBdfICb4N14+GARUDw2XV2N4tvzg==", "integrity": "sha512-PEIGCY5tSlUt50cqyMXfCzX+oOPqN0vuGqWzbcJ2xvnkzkq46oOpz7dQaTDBdfICb4N14+GARUDw2XV2N4tvzg==",
"dev": true,
"license": "MIT", "license": "MIT",
"engines": { "engines": {
"node": ">=10.0.0" "node": ">=10.0.0"
@ -30130,6 +30162,7 @@
"@fastify/helmet": "^13.0.2", "@fastify/helmet": "^13.0.2",
"@fastify/rate-limit": "^10.3.0", "@fastify/rate-limit": "^10.3.0",
"@fastify/static": "^8.3.0", "@fastify/static": "^8.3.0",
"@fastify/websocket": "^11.2.0",
"@prisma/client": "5.22.0", "@prisma/client": "5.22.0",
"@simplewebauthn/server": "^13.2.2", "@simplewebauthn/server": "^13.2.2",
"@simplewebauthn/types": "^12.0.0", "@simplewebauthn/types": "^12.0.0",
@ -30148,6 +30181,7 @@
"@types/node": "^18.0.0", "@types/node": "^18.0.0",
"@types/nodemailer": "^7.0.5", "@types/nodemailer": "^7.0.5",
"@types/supertest": "^6.0.3", "@types/supertest": "^6.0.3",
"@types/ws": "^8.18.1",
"nodemon": "^3.1.11", "nodemon": "^3.1.11",
"prisma": "5.22.0", "prisma": "5.22.0",
"supertest": "^7.1.4", "supertest": "^7.1.4",

View file

@ -1,19 +1,17 @@
# ---- Builder stage ---- # ---- Builder stage ----
FROM node:22-alpine AS builder FROM node:24-alpine AS builder
WORKDIR /repo WORKDIR /repo
RUN apk add --no-cache openssl libc6-compat RUN apk add --no-cache openssl libc6-compat
# Upgrade npm to match lockfile version (npm 11)
RUN npm install -g npm@11
# Root package + lock (for workspace install) # Root package + lock (for workspace install)
COPY package.json package-lock.json ./ COPY package.json package-lock.json ./
COPY packages/shared-schema/package.json ./packages/shared-schema/ COPY packages/shared-schema/package.json ./packages/shared-schema/
COPY packages/super-sync-server/package.json ./packages/super-sync-server/ COPY packages/super-sync-server/package.json ./packages/super-sync-server/
# Install ALL dependencies for the monorepo # Install ALL dependencies for the monorepo
# Use npm ci for reproducible builds; requires compatible lockfile
RUN npm ci --ignore-scripts RUN npm ci --ignore-scripts
# Copy source code # Copy source code
@ -32,7 +30,7 @@ RUN npx prisma generate
RUN rm -rf dist && npm run build RUN rm -rf dist && npm run build
# ---- Runtime stage ---- # ---- Runtime stage ----
FROM node:22-alpine AS production FROM node:24-alpine AS production
# System deps for Prisma & healthcheck + create user BEFORE copying files # System deps for Prisma & healthcheck + create user BEFORE copying files
RUN apk add --no-cache openssl libc6-compat wget && \ RUN apk add --no-cache openssl libc6-compat wget && \
@ -69,4 +67,4 @@ ENV DATA_DIR=/app/data
HEALTHCHECK --interval=30s --timeout=5s --start-period=10s --retries=3 \ HEALTHCHECK --interval=30s --timeout=5s --start-period=10s --retries=3 \
CMD wget --no-verbose --tries=1 --spider http://localhost:${PORT}/health || exit 1 CMD wget --no-verbose --tries=1 --spider http://localhost:${PORT}/health || exit 1
CMD ["sh", "-c", "npx prisma db push --skip-generate && node dist/src/index.js"] CMD ["sh", "-c", "npx prisma@5.22.0 migrate deploy && node dist/src/index.js"]

View file

@ -0,0 +1,8 @@
.git
.gitignore
*.swp
*.bak
.project
.idea/
*.tmproj
.vscode/

View file

@ -0,0 +1,12 @@
apiVersion: v2
name: supersync
description: Helm chart for SuperSync - Super Productivity sync server
type: application
version: 0.1.0
appVersion: '1.0.0'
home: https://super-productivity.com
sources:
- https://github.com/super-productivity/super-productivity
maintainers:
- name: johannesjo
url: https://github.com/johannesjo

View file

@ -0,0 +1,27 @@
SuperSync has been deployed!
{{- if not .Values.supersync.publicUrl }}
WARNING: supersync.publicUrl is not set. Server will reject requests in production mode.
{{- end }}
{{- if and .Values.ingress.enabled .Values.ingress.tls (eq (len .Values.ingress.tls) 0) }}
WARNING: Ingress is enabled without TLS. Consider enabling TLS for production use.
{{- end }}
{{- if and .Values.ingress.enabled (not .Values.ingress.tls) }}
WARNING: Ingress is enabled without TLS. Consider enabling TLS for production use.
{{- end }}
Get the application URL:
{{- if .Values.ingress.enabled }}
{{- range .Values.ingress.hosts }}
http{{ if $.Values.ingress.tls }}s{{ end }}://{{ .host }}
{{- end }}
{{- else if eq .Values.service.type "NodePort" }}
export NODE_PORT=$(kubectl get --namespace {{ .Release.Namespace }} -o jsonpath="{.spec.ports[0].nodePort}" services {{ include "supersync.fullname" . }})
export NODE_IP=$(kubectl get nodes --namespace {{ .Release.Namespace }} -o jsonpath="{.items[0].status.addresses[0].address}")
echo http://$NODE_IP:$NODE_PORT
{{- else }}
kubectl port-forward --namespace {{ .Release.Namespace }} svc/{{ include "supersync.fullname" . }} 8080:{{ .Values.service.port }}
echo http://localhost:8080
{{- end }}

View file

@ -0,0 +1,69 @@
{{/*
Expand the name of the chart.
*/}}
{{- define "supersync.name" -}}
{{- default .Chart.Name .Values.nameOverride | trunc 63 | trimSuffix "-" }}
{{- end }}
{{/*
Create a default fully qualified app name.
We truncate at 63 chars because some Kubernetes name fields are limited to this (by the DNS naming spec).
If release name contains chart name it will be used as a full name.
*/}}
{{- define "supersync.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 name and version as used by the chart label.
*/}}
{{- define "supersync.chart" -}}
{{- printf "%s-%s" .Chart.Name .Chart.Version | replace "+" "_" | trunc 63 | trimSuffix "-" }}
{{- end }}
{{/*
Common labels
*/}}
{{- define "supersync.labels" -}}
helm.sh/chart: {{ include "supersync.chart" . }}
{{ include "supersync.selectorLabels" . }}
{{- if .Chart.AppVersion }}
app.kubernetes.io/version: {{ .Chart.AppVersion | quote }}
{{- end }}
app.kubernetes.io/managed-by: {{ .Release.Service }}
{{- end }}
{{/*
Selector labels
*/}}
{{- define "supersync.selectorLabels" -}}
app.kubernetes.io/name: {{ include "supersync.name" . }}
app.kubernetes.io/instance: {{ .Release.Name }}
{{- end }}
{{/*
Create the name of the service account to use
*/}}
{{- define "supersync.serviceAccountName" -}}
{{- if .Values.serviceAccount.create }}
{{- default (include "supersync.fullname" .) .Values.serviceAccount.name }}
{{- else }}
{{- default "default" .Values.serviceAccount.name }}
{{- end }}
{{- end }}
{{/*
Create the fully qualified name for the PostgreSQL resources.
*/}}
{{- define "supersync.postgresql.fullname" -}}
{{- printf "%s-postgresql" (include "supersync.fullname" .) | trunc 63 | trimSuffix "-" }}
{{- end }}

View file

@ -0,0 +1,35 @@
apiVersion: v1
kind: ConfigMap
metadata:
name: {{ include "supersync.fullname" . }}
labels:
{{- include "supersync.labels" . | nindent 4 }}
data:
PORT: {{ .Values.supersync.port | quote }}
NODE_ENV: {{ .Values.supersync.nodeEnv | quote }}
{{- if .Values.supersync.publicUrl }}
PUBLIC_URL: {{ .Values.supersync.publicUrl | quote }}
{{- end }}
CORS_ORIGINS: {{ .Values.supersync.cors.origins | quote }}
{{- if .Values.supersync.webauthn.rpName }}
WEBAUTHN_RP_NAME: {{ .Values.supersync.webauthn.rpName | quote }}
{{- end }}
{{- if .Values.supersync.webauthn.rpId }}
WEBAUTHN_RP_ID: {{ .Values.supersync.webauthn.rpId | quote }}
{{- end }}
{{- if .Values.supersync.webauthn.origin }}
WEBAUTHN_ORIGIN: {{ .Values.supersync.webauthn.origin | quote }}
{{- end }}
{{- if .Values.supersync.smtp.enabled }}
SMTP_HOST: {{ .Values.supersync.smtp.host | quote }}
SMTP_PORT: {{ .Values.supersync.smtp.port | quote }}
SMTP_SECURE: {{ .Values.supersync.smtp.secure | quote }}
SMTP_FROM: {{ .Values.supersync.smtp.from | quote }}
{{- end }}
{{- if .Values.supersync.privacy.contactName }}
PRIVACY_CONTACT_NAME: {{ .Values.supersync.privacy.contactName | quote }}
PRIVACY_ADDRESS_STREET: {{ .Values.supersync.privacy.addressStreet | quote }}
PRIVACY_ADDRESS_CITY: {{ .Values.supersync.privacy.addressCity | quote }}
PRIVACY_ADDRESS_COUNTRY: {{ .Values.supersync.privacy.addressCountry | quote }}
PRIVACY_CONTACT_EMAIL: {{ .Values.supersync.privacy.contactEmail | quote }}
{{- end }}

View file

@ -0,0 +1,187 @@
{{- if gt (.Values.replicaCount | int) 1 }}
{{- fail "replicaCount > 1 is not supported: WebSocket connection state is in-memory and not shared across replicas" }}
{{- end }}
apiVersion: apps/v1
kind: Deployment
metadata:
name: {{ include "supersync.fullname" . }}
labels:
{{- include "supersync.labels" . | nindent 4 }}
spec:
{{- if not .Values.autoscaling.enabled }}
replicas: {{ .Values.replicaCount }}
{{- end }}
{{- if .Values.supersync.persistence.enabled }}
strategy:
type: Recreate
{{- end }}
selector:
matchLabels:
{{- include "supersync.selectorLabels" . | nindent 6 }}
template:
metadata:
annotations:
checksum/config: {{ include (print $.Template.BasePath "/configmap.yaml") . | sha256sum }}
checksum/secret: {{ include (print $.Template.BasePath "/secret.yaml") . | sha256sum }}
{{- with .Values.podAnnotations }}
{{- toYaml . | nindent 8 }}
{{- end }}
labels:
{{- include "supersync.labels" . | nindent 8 }}
{{- with .Values.podLabels }}
{{- toYaml . | nindent 8 }}
{{- end }}
spec:
{{- with .Values.imagePullSecrets }}
imagePullSecrets:
{{- toYaml . | nindent 8 }}
{{- end }}
serviceAccountName: {{ include "supersync.serviceAccountName" . }}
securityContext:
{{- toYaml .Values.podSecurityContext | nindent 8 }}
initContainers:
- name: migrate-db
image: "{{ .Values.image.repository }}:{{ .Values.image.tag | default .Chart.AppVersion }}"
imagePullPolicy: {{ .Values.image.pullPolicy }}
command: ['sh', '-c', 'npx prisma@5.22.0 migrate deploy']
env:
{{- if .Values.postgresql.enabled }}
- name: POSTGRES_USER
value: {{ .Values.postgresql.username | quote }}
- name: POSTGRES_DB
value: {{ .Values.postgresql.database | quote }}
- name: POSTGRES_PASSWORD
valueFrom:
secretKeyRef:
name: {{ .Values.postgresql.existingSecret | default (include "supersync.postgresql.fullname" .) }}
key: POSTGRES_PASSWORD
- name: DATABASE_URL
value: "postgresql://$(POSTGRES_USER):$(POSTGRES_PASSWORD)@{{ include "supersync.postgresql.fullname" . }}:5432/$(POSTGRES_DB)"
{{- else }}
- name: DATABASE_URL
{{- if .Values.externalDatabase.existingSecret }}
valueFrom:
secretKeyRef:
name: {{ .Values.externalDatabase.existingSecret }}
key: DATABASE_URL
{{- else }}
value: {{ .Values.externalDatabase.url | quote }}
{{- end }}
{{- end }}
securityContext:
{{- toYaml .Values.securityContext | nindent 12 }}
- name: copy-public
image: "{{ .Values.image.repository }}:{{ .Values.image.tag | default .Chart.AppVersion }}"
imagePullPolicy: {{ .Values.image.pullPolicy }}
command: ['sh', '-c', 'cp -r /app/public/* /tmp/public/']
volumeMounts:
- name: public
mountPath: /tmp/public
securityContext:
{{- toYaml .Values.securityContext | nindent 12 }}
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.supersync.port }}
protocol: TCP
envFrom:
- configMapRef:
name: {{ include "supersync.fullname" . }}
env:
- name: JWT_SECRET
valueFrom:
secretKeyRef:
name: {{ .Values.supersync.existingSecret | default (include "supersync.fullname" .) }}
key: JWT_SECRET
{{- if .Values.postgresql.enabled }}
- name: POSTGRES_USER
value: {{ .Values.postgresql.username | quote }}
- name: POSTGRES_DB
value: {{ .Values.postgresql.database | quote }}
- name: POSTGRES_PASSWORD
valueFrom:
secretKeyRef:
name: {{ .Values.postgresql.existingSecret | default (include "supersync.postgresql.fullname" .) }}
key: POSTGRES_PASSWORD
- name: DATABASE_URL
value: "postgresql://$(POSTGRES_USER):$(POSTGRES_PASSWORD)@{{ include "supersync.postgresql.fullname" . }}:5432/$(POSTGRES_DB)"
{{- else }}
- name: DATABASE_URL
{{- if .Values.externalDatabase.existingSecret }}
valueFrom:
secretKeyRef:
name: {{ .Values.externalDatabase.existingSecret }}
key: DATABASE_URL
{{- else }}
value: {{ .Values.externalDatabase.url | quote }}
{{- end }}
{{- end }}
{{- if .Values.supersync.smtp.enabled }}
- name: SMTP_USER
valueFrom:
secretKeyRef:
name: {{ .Values.supersync.smtp.existingSecret | default (include "supersync.fullname" .) }}
key: SMTP_USER
- name: SMTP_PASS
valueFrom:
secretKeyRef:
name: {{ .Values.supersync.smtp.existingSecret | default (include "supersync.fullname" .) }}
key: SMTP_PASS
{{- end }}
startupProbe:
httpGet:
path: /health
port: http
initialDelaySeconds: 10
periodSeconds: 5
failureThreshold: 30
livenessProbe:
httpGet:
path: /health
port: http
periodSeconds: 30
timeoutSeconds: 5
readinessProbe:
httpGet:
path: /health
port: http
periodSeconds: 10
timeoutSeconds: 5
resources:
{{- toYaml .Values.resources | nindent 12 }}
volumeMounts:
- name: tmp
mountPath: /tmp
- name: public
mountPath: /app/public
- name: data
mountPath: /app/data
volumes:
- name: tmp
emptyDir: {}
- name: public
emptyDir: {}
- name: data
{{- if .Values.supersync.persistence.enabled }}
persistentVolumeClaim:
claimName: {{ include "supersync.fullname" . }}
{{- else }}
emptyDir: {}
{{- end }}
{{- with .Values.nodeSelector }}
nodeSelector:
{{- toYaml . | nindent 8 }}
{{- end }}
{{- with .Values.affinity }}
affinity:
{{- toYaml . | nindent 8 }}
{{- end }}
{{- with .Values.tolerations }}
tolerations:
{{- toYaml . | nindent 8 }}
{{- end }}

View file

@ -0,0 +1,25 @@
{{- if and .Values.autoscaling.enabled (gt (.Values.autoscaling.maxReplicas | int) 1) }}
{{- fail "autoscaling with maxReplicas > 1 is not supported: WebSocket connection state is in-memory and not shared across replicas" }}
{{- end }}
{{- if .Values.autoscaling.enabled -}}
apiVersion: autoscaling/v2
kind: HorizontalPodAutoscaler
metadata:
name: {{ include "supersync.fullname" . }}
labels:
{{- include "supersync.labels" . | nindent 4 }}
spec:
scaleTargetRef:
apiVersion: apps/v1
kind: Deployment
name: {{ include "supersync.fullname" . }}
minReplicas: {{ .Values.autoscaling.minReplicas }}
maxReplicas: {{ .Values.autoscaling.maxReplicas }}
metrics:
- type: Resource
resource:
name: cpu
target:
type: Utilization
averageUtilization: {{ .Values.autoscaling.targetCPUUtilizationPercentage }}
{{- end }}

View file

@ -0,0 +1,41 @@
{{- if .Values.ingress.enabled -}}
apiVersion: networking.k8s.io/v1
kind: Ingress
metadata:
name: {{ include "supersync.fullname" . }}
labels:
{{- include "supersync.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:
{{- range .Values.ingress.tls }}
- hosts:
{{- range .hosts }}
- {{ . | quote }}
{{- end }}
secretName: {{ .secretName }}
{{- end }}
{{- end }}
rules:
{{- range .Values.ingress.hosts }}
- host: {{ .host | quote }}
http:
paths:
{{- range .paths }}
- path: {{ .path }}
pathType: {{ .pathType }}
backend:
service:
name: {{ include "supersync.fullname" $ }}
port:
name: http
{{- end }}
{{- end }}
{{- end }}

View file

@ -0,0 +1,49 @@
{{- if .Values.networkPolicy.enabled -}}
apiVersion: networking.k8s.io/v1
kind: NetworkPolicy
metadata:
name: {{ include "supersync.fullname" . }}
labels:
{{- include "supersync.labels" . | nindent 4 }}
spec:
podSelector:
matchLabels:
{{- include "supersync.selectorLabels" . | nindent 6 }}
policyTypes:
- Ingress
- Egress
ingress:
- from:
- namespaceSelector:
matchLabels:
kubernetes.io/metadata.name: {{ .Values.networkPolicy.ingressNamespace | default "ingress-nginx" }}
ports:
- port: {{ .Values.supersync.port }}
protocol: TCP
egress:
{{- if .Values.postgresql.enabled }}
- to:
- podSelector:
matchLabels:
app.kubernetes.io/name: {{ include "supersync.name" . }}-postgresql
app.kubernetes.io/instance: {{ .Release.Name }}
ports:
- port: 5432
protocol: TCP
{{- else }}
# Allow egress to external database (any destination on port 5432)
- ports:
- port: 5432
protocol: TCP
{{- end }}
- ports:
- port: 53
protocol: TCP
- port: 53
protocol: UDP
{{- if .Values.supersync.smtp.enabled }}
- ports:
- port: {{ .Values.supersync.smtp.port }}
protocol: TCP
{{- end }}
{{- end }}

View file

@ -0,0 +1,13 @@
{{- if .Values.podDisruptionBudget.enabled -}}
apiVersion: policy/v1
kind: PodDisruptionBudget
metadata:
name: {{ include "supersync.fullname" . }}
labels:
{{- include "supersync.labels" . | nindent 4 }}
spec:
maxUnavailable: {{ .Values.podDisruptionBudget.maxUnavailable | default 1 }}
selector:
matchLabels:
{{- include "supersync.selectorLabels" . | nindent 6 }}
{{- end }}

View file

@ -0,0 +1,14 @@
{{- if and .Values.postgresql.enabled (not .Values.postgresql.existingSecret) (eq .Values.postgresql.password "") }}
{{- fail "postgresql.password is required when postgresql.enabled is true and postgresql.existingSecret is not set" }}
{{- end }}
{{- if and .Values.postgresql.enabled (not .Values.postgresql.existingSecret) -}}
apiVersion: v1
kind: Secret
metadata:
name: {{ include "supersync.postgresql.fullname" . }}
labels:
{{- include "supersync.labels" . | nindent 4 }}
type: Opaque
stringData:
POSTGRES_PASSWORD: {{ .Values.postgresql.password | quote }}
{{- end }}

View file

@ -0,0 +1,19 @@
{{- if .Values.postgresql.enabled -}}
apiVersion: v1
kind: Service
metadata:
name: {{ include "supersync.postgresql.fullname" . }}
labels:
{{- include "supersync.labels" . | nindent 4 }}
app.kubernetes.io/component: postgresql
spec:
type: ClusterIP
ports:
- port: 5432
targetPort: postgresql
protocol: TCP
name: postgresql
selector:
app.kubernetes.io/name: {{ include "supersync.name" . }}-postgresql
app.kubernetes.io/instance: {{ .Release.Name }}
{{- end }}

View file

@ -0,0 +1,97 @@
{{- if .Values.postgresql.enabled -}}
apiVersion: apps/v1
kind: StatefulSet
metadata:
name: {{ include "supersync.postgresql.fullname" . }}
labels:
{{- include "supersync.labels" . | nindent 4 }}
app.kubernetes.io/component: postgresql
spec:
serviceName: {{ include "supersync.postgresql.fullname" . }}
replicas: 1
selector:
matchLabels:
app.kubernetes.io/name: {{ include "supersync.name" . }}-postgresql
app.kubernetes.io/instance: {{ .Release.Name }}
template:
metadata:
labels:
app.kubernetes.io/name: {{ include "supersync.name" . }}-postgresql
app.kubernetes.io/instance: {{ .Release.Name }}
app.kubernetes.io/component: postgresql
spec:
securityContext:
runAsUser: 999
fsGroup: 999
containers:
- name: postgresql
image: "{{ .Values.postgresql.image.repository }}:{{ .Values.postgresql.image.tag }}"
imagePullPolicy: {{ .Values.postgresql.image.pullPolicy }}
securityContext:
allowPrivilegeEscalation: false
capabilities:
drop:
- ALL
ports:
- name: postgresql
containerPort: 5432
protocol: TCP
env:
- name: POSTGRES_USER
value: {{ .Values.postgresql.username | quote }}
- name: POSTGRES_DB
value: {{ .Values.postgresql.database | quote }}
- name: POSTGRES_PASSWORD
valueFrom:
secretKeyRef:
name: {{ .Values.postgresql.existingSecret | default (include "supersync.postgresql.fullname" .) }}
key: POSTGRES_PASSWORD
- name: PGDATA
value: /var/lib/postgresql/data/pgdata
readinessProbe:
exec:
command:
- pg_isready
- -U
- {{ .Values.postgresql.username }}
- -d
- {{ .Values.postgresql.database }}
periodSeconds: 10
timeoutSeconds: 5
livenessProbe:
exec:
command:
- pg_isready
- -U
- {{ .Values.postgresql.username }}
- -d
- {{ .Values.postgresql.database }}
periodSeconds: 30
timeoutSeconds: 5
resources:
{{- toYaml .Values.postgresql.resources | nindent 12 }}
volumeMounts:
- name: data
mountPath: /var/lib/postgresql/data
{{- if not .Values.postgresql.persistence.enabled }}
volumes:
- name: data
emptyDir: {}
{{- end }}
{{- if .Values.postgresql.persistence.enabled }}
volumeClaimTemplates:
- metadata:
name: data
spec:
accessModes:
{{- range .Values.postgresql.persistence.accessModes }}
- {{ . }}
{{- end }}
{{- if .Values.postgresql.persistence.storageClass }}
storageClassName: {{ .Values.postgresql.persistence.storageClass | quote }}
{{- end }}
resources:
requests:
storage: {{ .Values.postgresql.persistence.size }}
{{- end }}
{{- end }}

View file

@ -0,0 +1,19 @@
{{- if .Values.supersync.persistence.enabled -}}
apiVersion: v1
kind: PersistentVolumeClaim
metadata:
name: {{ include "supersync.fullname" . }}
labels:
{{- include "supersync.labels" . | nindent 4 }}
spec:
accessModes:
{{- range .Values.supersync.persistence.accessModes }}
- {{ . }}
{{- end }}
{{- if .Values.supersync.persistence.storageClass }}
storageClassName: {{ .Values.supersync.persistence.storageClass | quote }}
{{- end }}
resources:
requests:
storage: {{ .Values.supersync.persistence.size }}
{{- end }}

View file

@ -0,0 +1,18 @@
{{- if and (not .Values.supersync.existingSecret) (eq .Values.supersync.secret.jwtSecret "") }}
{{- fail "supersync.secret.jwtSecret is required (min 32 chars) when supersync.existingSecret is not set" }}
{{- end }}
{{- if not .Values.supersync.existingSecret -}}
apiVersion: v1
kind: Secret
metadata:
name: {{ include "supersync.fullname" . }}
labels:
{{- include "supersync.labels" . | nindent 4 }}
type: Opaque
stringData:
JWT_SECRET: {{ .Values.supersync.secret.jwtSecret | quote }}
{{- if and .Values.supersync.smtp.enabled (not .Values.supersync.smtp.existingSecret) }}
SMTP_USER: {{ .Values.supersync.smtp.user | quote }}
SMTP_PASS: {{ .Values.supersync.smtp.password | quote }}
{{- end }}
{{- end }}

View file

@ -0,0 +1,15 @@
apiVersion: v1
kind: Service
metadata:
name: {{ include "supersync.fullname" . }}
labels:
{{- include "supersync.labels" . | nindent 4 }}
spec:
type: {{ .Values.service.type }}
ports:
- port: {{ .Values.service.port }}
targetPort: {{ .Values.supersync.port }}
protocol: TCP
name: http
selector:
{{- include "supersync.selectorLabels" . | nindent 4 }}

View file

@ -0,0 +1,13 @@
{{- if .Values.serviceAccount.create -}}
apiVersion: v1
kind: ServiceAccount
metadata:
name: {{ include "supersync.serviceAccountName" . }}
labels:
{{- include "supersync.labels" . | nindent 4 }}
{{- with .Values.serviceAccount.annotations }}
annotations:
{{- toYaml . | nindent 4 }}
{{- end }}
automountServiceAccountToken: {{ .Values.serviceAccount.automount }}
{{- end }}

View file

@ -0,0 +1,14 @@
apiVersion: v1
kind: Pod
metadata:
name: "{{ include "supersync.fullname" . }}-test-connection"
labels:
{{- include "supersync.labels" . | nindent 4 }}
annotations:
"helm.sh/hook": test
spec:
restartPolicy: Never
containers:
- name: wget
image: busybox:1.36
command: ['wget', '--spider', '--timeout=5', 'http://{{ include "supersync.fullname" . }}:{{ .Values.service.port }}/health']

View file

@ -0,0 +1,228 @@
# -- Number of SuperSync replicas.
# Running more than 1 replica requires a shared store (e.g., Redis) for in-memory caches and WebSocket connection state (not yet supported).
replicaCount: 1
image:
# -- SuperSync container image repository
repository: ghcr.io/super-productivity/supersync
# -- Image pull policy
pullPolicy: IfNotPresent
# -- Overrides the image tag. Defaults to the chart appVersion.
tag: ""
# -- Docker registry pull secrets
imagePullSecrets: []
# -- Override the release name prefix
nameOverride: ""
# -- Override the full release name
fullnameOverride: ""
serviceAccount:
# -- Whether to create a ServiceAccount
create: true
# -- Automount the service account token (not needed by this app)
automount: false
# -- Annotations to add to the ServiceAccount
annotations: {}
# -- The name of the ServiceAccount to use. If not set and create is true, a name is generated.
name: ""
# -- SuperSync application configuration
supersync:
# -- Application listening port
port: 1900
# -- Node.js environment
nodeEnv: production
# -- REQUIRED: Public URL of the sync server (e.g. https://sync.example.com)
publicUrl: ""
cors:
# -- Comma-separated list of allowed CORS origins
origins: "https://app.super-productivity.com"
webauthn:
# -- WebAuthn relying party display name
rpName: "Super Productivity Sync"
# -- WebAuthn relying party ID (e.g. sync.example.com)
rpId: ""
# -- WebAuthn origin (e.g. https://sync.example.com)
origin: ""
smtp:
# -- Enable SMTP email sending
enabled: false
# -- SMTP server hostname
host: ""
# -- SMTP server port
port: 587
# -- Use TLS for SMTP
secure: false
# -- Sender email address
from: ""
# -- SMTP username
user: ""
# -- SMTP password
password: ""
# -- Use an existing Kubernetes secret for SMTP credentials.
# The secret must contain SMTP_USER and SMTP_PASS keys.
existingSecret: ""
privacy:
# -- Contact name for privacy policy
contactName: ""
# -- Street address for privacy policy
addressStreet: ""
# -- City for privacy policy
addressCity: ""
# -- Country for privacy policy
addressCountry: ""
# -- Contact email for privacy policy
contactEmail: ""
persistence:
# -- Enable persistent storage for /app/data
enabled: true
# -- Size of the persistent volume
size: 1Gi
# -- Storage class for the PVC. Empty string uses the default storage class.
storageClass: ""
# -- Access modes for the PVC
accessModes:
- ReadWriteOnce
secret:
# -- REQUIRED: JWT signing secret (minimum 32 characters)
jwtSecret: ""
# -- Use a pre-existing Kubernetes secret instead of creating one.
# The secret must contain a JWT_SECRET key.
existingSecret: ""
# -- Built-in PostgreSQL configuration
postgresql:
# -- Deploy a PostgreSQL instance alongside SuperSync.
# Set to false and configure externalDatabase to use an external PostgreSQL.
enabled: true
image:
# -- PostgreSQL image repository
repository: postgres
# -- PostgreSQL image tag
tag: "16-alpine"
# -- PostgreSQL image pull policy
pullPolicy: IfNotPresent
# -- PostgreSQL database name
database: supersync
# -- PostgreSQL username
username: supersync
# -- PostgreSQL password (REQUIRED when postgresql.enabled is true)
password: ""
# -- Use an existing Kubernetes secret for the PostgreSQL password.
# The secret must contain a POSTGRES_PASSWORD key.
existingSecret: ""
persistence:
# -- Enable persistent storage for PostgreSQL data
enabled: true
# -- Size of the PostgreSQL persistent volume
size: 5Gi
# -- Storage class for the PostgreSQL PVC. Empty string uses the default.
storageClass: ""
# -- Access modes for the PostgreSQL PVC
accessModes:
- ReadWriteOnce
# -- Resource requests and limits for PostgreSQL
resources:
requests:
cpu: 100m
memory: 256Mi
limits:
memory: 512Mi
# -- External database configuration (used when postgresql.enabled is false)
externalDatabase:
# -- Full PostgreSQL connection URL (e.g. postgresql://user:pass@host:5432/dbname)
url: ""
# -- Use an existing Kubernetes secret for the database URL.
# The secret must contain a DATABASE_URL key.
existingSecret: ""
# -- Ingress configuration
ingress:
# -- Enable ingress
enabled: false
# -- Ingress class name
className: ""
# -- Ingress annotations
# WebSocket support requires extended timeouts. Example for nginx-ingress:
# nginx.ingress.kubernetes.io/proxy-read-timeout: "3600"
# nginx.ingress.kubernetes.io/proxy-send-timeout: "3600"
# For Traefik:
# traefik.ingress.kubernetes.io/router.middlewares: ...
annotations: {}
# -- Ingress host rules
hosts:
- host: sync.example.com
paths:
- path: /
pathType: Prefix
# -- Ingress TLS configuration
tls: []
# - secretName: supersync-tls
# hosts:
# - sync.example.com
# -- Service configuration
service:
# -- Service type
type: ClusterIP
# -- Service port
port: 80
# -- Resource requests and limits for the SuperSync container
resources:
requests:
cpu: 100m
memory: 128Mi
limits:
memory: 256Mi
# -- Horizontal Pod Autoscaler configuration
autoscaling:
# -- Enable HPA
enabled: false
# -- Minimum number of replicas
minReplicas: 1
# -- Maximum number of replicas (must be 1 until shared WS state is implemented)
maxReplicas: 1
# -- Target CPU utilization percentage
targetCPUUtilizationPercentage: 80
# -- Pod Disruption Budget configuration
podDisruptionBudget:
# -- Enable PDB
enabled: false
# -- Maximum number of unavailable pods (use this instead of minAvailable for single-replica)
maxUnavailable: 1
# -- Network Policy configuration
networkPolicy:
# -- Enable NetworkPolicy
enabled: false
# -- Namespace of the ingress controller (for ingress from: selector)
ingressNamespace: "ingress-nginx"
# -- Node selector for pod scheduling
nodeSelector: {}
# -- Tolerations for pod scheduling
tolerations: []
# -- Affinity rules for pod scheduling
affinity: {}
# -- Annotations to add to pods
podAnnotations: {}
# -- Labels to add to pods
podLabels: {}
# -- Pod-level security context
podSecurityContext:
fsGroup: 1001
# -- Container-level security context
securityContext:
runAsUser: 1001
runAsGroup: 1001
runAsNonRoot: true
readOnlyRootFilesystem: true
allowPrivilegeEscalation: false
capabilities:
drop:
- ALL

View file

@ -34,6 +34,7 @@
"@fastify/helmet": "^13.0.2", "@fastify/helmet": "^13.0.2",
"@fastify/rate-limit": "^10.3.0", "@fastify/rate-limit": "^10.3.0",
"@fastify/static": "^8.3.0", "@fastify/static": "^8.3.0",
"@fastify/websocket": "^11.2.0",
"@prisma/client": "5.22.0", "@prisma/client": "5.22.0",
"@simplewebauthn/server": "^13.2.2", "@simplewebauthn/server": "^13.2.2",
"@simplewebauthn/types": "^12.0.0", "@simplewebauthn/types": "^12.0.0",
@ -52,6 +53,7 @@
"@types/node": "^18.0.0", "@types/node": "^18.0.0",
"@types/nodemailer": "^7.0.5", "@types/nodemailer": "^7.0.5",
"@types/supertest": "^6.0.3", "@types/supertest": "^6.0.3",
"@types/ws": "^8.18.1",
"nodemon": "^3.1.11", "nodemon": "^3.1.11",
"prisma": "5.22.0", "prisma": "5.22.0",
"supertest": "^7.1.4", "supertest": "^7.1.4",

View file

@ -1,5 +1,6 @@
import { FastifyInstance } from 'fastify'; import { FastifyInstance } from 'fastify';
import { z } from 'zod'; import { z } from 'zod';
import { isEmailAllowed } from './email-allowlist';
import * as jwt from 'jsonwebtoken'; import * as jwt from 'jsonwebtoken';
import { import {
verifyEmail, verifyEmail,
@ -246,6 +247,10 @@ export const apiRoutes = async (fastify: FastifyInstance): Promise<void> => {
} }
const { email } = parseResult.data; const { email } = parseResult.data;
if (!isEmailAllowed(email)) {
return reply.status(403).send({ error: 'Registration is not allowed for this email address.' });
}
const options = await generateRegistrationOptions(email); const options = await generateRegistrationOptions(email);
return reply.send(options); return reply.send(options);
} catch (err) { } catch (err) {
@ -280,6 +285,10 @@ export const apiRoutes = async (fastify: FastifyInstance): Promise<void> => {
} }
const { email, credential } = parseResult.data; const { email, credential } = parseResult.data;
if (!isEmailAllowed(email)) {
return reply.status(403).send({ error: 'Registration is not allowed for this email address.' });
}
const result = await verifyRegistration(email, credential as any, Date.now()); const result = await verifyRegistration(email, credential as any, Date.now());
return reply.status(201).send(result); return reply.status(201).send(result);
} catch (err) { } catch (err) {
@ -509,6 +518,10 @@ export const apiRoutes = async (fastify: FastifyInstance): Promise<void> => {
} }
const { email } = parseResult.data; const { email } = parseResult.data;
if (!isEmailAllowed(email)) {
return reply.status(403).send({ error: 'Registration is not allowed for this email address.' });
}
const result = await registerWithMagicLink(email, Date.now()); const result = await registerWithMagicLink(email, Date.now());
return reply.status(201).send(result); return reply.status(201).send(result);
} catch (err) { } catch (err) {

View file

@ -0,0 +1,30 @@
/**
* Email allowlist for restricting registration.
*
* Set ALLOWED_EMAILS env var to a comma-separated list of:
* - Fully qualified emails: user@example.com
* - Domain wildcards: *@example.com
*
* When unset, all emails are allowed (open registration).
*/
import { Logger } from './logger';
const rules: string[] = (process.env.ALLOWED_EMAILS ?? '')
.split(',')
.map((e) => e.trim().toLowerCase())
.filter((e) => e.length > 0);
if (rules.length > 0) {
Logger.info(`Email allowlist enabled: ${rules.length} rule(s)`);
}
export const isEmailAllowed = (email: string): boolean => {
if (rules.length === 0) return true;
const normalized = email.toLowerCase();
const domain = normalized.split('@')[1];
return rules.some((rule) =>
rule.startsWith('*@') ? domain === rule.slice(2) : normalized === rule,
);
};

View file

@ -8,9 +8,15 @@ import * as path from 'path';
import { loadConfigFromEnv, ServerConfig, PrivacyConfig } from './config'; import { loadConfigFromEnv, ServerConfig, PrivacyConfig } from './config';
import { Logger } from './logger'; import { Logger } from './logger';
import { prisma, disconnectDb } from './db'; import { prisma, disconnectDb } from './db';
import websocket from '@fastify/websocket';
import { apiRoutes } from './api'; import { apiRoutes } from './api';
import { pageRoutes } from './pages'; import { pageRoutes } from './pages';
import { syncRoutes, startCleanupJobs, stopCleanupJobs } from './sync'; import { syncRoutes, startCleanupJobs, stopCleanupJobs } from './sync';
import { wsRoutes } from './sync/websocket.routes';
import {
getWsConnectionService,
resetWsConnectionService,
} from './sync/services/websocket-connection.service';
import { testRoutes } from './test-routes'; import { testRoutes } from './test-routes';
// HTML escape to prevent XSS in generated HTML // HTML escape to prevent XSS in generated HTML
@ -151,12 +157,20 @@ export const createServer = (
prefix: '/', prefix: '/',
}); });
// WebSocket support for real-time sync notifications
// maxPayload: only app-level pong messages expected from clients (~20 bytes)
await fastifyServer.register(websocket, {
options: { maxPayload: 1024 },
});
// Health Check - verifies database connectivity // Health Check - verifies database connectivity
fastifyServer.get('/health', async (_, reply) => { // Exempt from rate limiting (Kubernetes probes hit this every 5-15s)
fastifyServer.get('/health', { config: { rateLimit: false } }, async (_, reply) => {
try { try {
// Simple query to verify DB is responsive // Simple query to verify DB is responsive
await prisma.$queryRaw`SELECT 1`; await prisma.$queryRaw`SELECT 1`;
return { status: 'ok', db: 'connected' }; const wsConnections = getWsConnectionService().getConnectionCount();
return { status: 'ok', db: 'connected', wsConnections };
} catch (err) { } catch (err) {
Logger.error('Health check failed: DB not responsive', err); Logger.error('Health check failed: DB not responsive', err);
return reply.status(503).send({ return reply.status(503).send({
@ -173,6 +187,9 @@ export const createServer = (
// Sync Routes (operation-based sync) // Sync Routes (operation-based sync)
await fastifyServer.register(syncRoutes, { prefix: '/api/sync' }); await fastifyServer.register(syncRoutes, { prefix: '/api/sync' });
// WebSocket routes for real-time sync notifications
await fastifyServer.register(wsRoutes, { prefix: '/api/sync' });
// Test Routes (only in test mode) // Test Routes (only in test mode)
if (fullConfig.testMode?.enabled) { if (fullConfig.testMode?.enabled) {
await fastifyServer.register(testRoutes, { prefix: '/api/test' }); await fastifyServer.register(testRoutes, { prefix: '/api/test' });
@ -185,6 +202,9 @@ export const createServer = (
// Start cleanup jobs // Start cleanup jobs
startCleanupJobs(); startCleanupJobs();
// Start WebSocket heartbeat
getWsConnectionService().startHeartbeat();
try { try {
const address = await fastifyServer.listen({ const address = await fastifyServer.listen({
port: fullConfig.port, port: fullConfig.port,
@ -198,6 +218,8 @@ export const createServer = (
} }
}, },
stop: async (): Promise<void> => { stop: async (): Promise<void> => {
// Stop WebSocket connections
resetWsConnectionService();
stopCleanupJobs(); stopCleanupJobs();
if (fastifyServer) { if (fastifyServer) {
await fastifyServer.close(); await fastifyServer.close();

View file

@ -1,4 +1,5 @@
export { syncRoutes } from './sync.routes'; export { syncRoutes } from './sync.routes';
export { wsRoutes } from './websocket.routes';
export { SyncService, getSyncService, initSyncService } from './sync.service'; export { SyncService, getSyncService, initSyncService } from './sync.service';
export { startCleanupJobs, stopCleanupJobs } from './cleanup'; export { startCleanupJobs, stopCleanupJobs } from './cleanup';
export * from './sync.types'; export * from './sync.types';

View file

@ -0,0 +1,291 @@
import { WebSocket } from 'ws';
import { Logger } from '../../logger';
interface ConnectedClient {
ws: WebSocket;
clientId: string;
userId: number;
lastPong: number;
}
/**
* Manages WebSocket connections for real-time sync notifications.
*
* Sends lightweight notifications when new operations are available,
* prompting clients to download via the existing HTTP endpoint.
* Does NOT stream operation payloads over WebSocket.
*/
export class WebSocketConnectionService {
private connections = new Map<number, Set<ConnectedClient>>();
private heartbeatInterval: ReturnType<typeof setInterval> | null = null;
/** 30s ping interval - keeps connection alive through proxies (most: 60-120s timeout) */
private static readonly PING_INTERVAL_MS = 30_000;
/** Close connection if no pong within 10s of ping */
private static readonly PONG_TIMEOUT_MS = 10_000;
/** Debounce notifications: max 1 per 100ms per user (latest-seq-wins) */
private static readonly NOTIFY_DEBOUNCE_MS = 100;
/** Max WebSocket connections per user to prevent resource exhaustion */
private static readonly MAX_CONNECTIONS_PER_USER = 10;
private pendingNotifications = new Map<
number,
{
timer: ReturnType<typeof setTimeout>;
excludeClientIds: Set<string>;
latestSeq: number;
}
>();
addConnection(userId: number, clientId: string, ws: WebSocket): void {
if (!this.connections.has(userId)) {
this.connections.set(userId, new Set());
}
const userSet = this.connections.get(userId)!;
if (userSet.size >= WebSocketConnectionService.MAX_CONNECTIONS_PER_USER) {
Logger.warn(
`[ws:user:${userId}] Connection rejected: max connections per user reached`,
);
ws.close(4008, 'Too many connections');
return;
}
const client: ConnectedClient = {
ws,
clientId,
userId,
lastPong: Date.now(),
};
userSet.add(client);
// Send connected message
this._sendMessage(ws, {
type: 'connected',
userId,
timestamp: Date.now(),
});
ws.on('pong', () => {
client.lastPong = Date.now();
});
ws.on('message', (data: Buffer) => {
try {
const msg = JSON.parse(data.toString());
if (msg.type === 'pong') {
client.lastPong = Date.now();
}
} catch (err) {
Logger.debug(`[ws:user:${userId}:${clientId}] Non-JSON message received`, err);
}
});
ws.on('close', () => {
this.removeConnection(userId, client);
});
ws.on('error', (err: Error) => {
// Close event follows error — cleanup is handled there
Logger.warn(`[ws:user:${userId}:${clientId}] WebSocket error: ${err.message}`);
});
const userConns = this.connections.get(userId)?.size ?? 0;
Logger.info(
`[ws:user:${userId}:${clientId}] Connected (${userConns} total for user)`,
);
}
removeConnection(userId: number, client: ConnectedClient): void {
const userSet = this.connections.get(userId);
if (userSet) {
userSet.delete(client);
if (userSet.size === 0) {
this.connections.delete(userId);
}
}
// Close the WebSocket if still open
if (
client.ws.readyState === WebSocket.OPEN ||
client.ws.readyState === WebSocket.CONNECTING
) {
try {
client.ws.close();
} catch (err) {
Logger.debug(
`[ws:user:${userId}:${client.clientId}] Error closing connection`,
err,
);
}
}
}
/**
* Notify all connected clients of a user (except the sender) about new operations.
* Uses debouncing to prevent notification storms during rapid uploads.
* Fire-and-forget - does not block the caller.
*/
notifyNewOps(userId: number, excludeClientId: string, latestSeq: number): void {
const userSet = this.connections.get(userId);
if (!userSet || userSet.size === 0) return;
let pending = this.pendingNotifications.get(userId);
if (pending) {
clearTimeout(pending.timer);
pending.excludeClientIds.add(excludeClientId);
pending.latestSeq = latestSeq;
} else {
pending = {
timer: null as unknown as ReturnType<typeof setTimeout>,
excludeClientIds: new Set([excludeClientId]),
latestSeq,
};
this.pendingNotifications.set(userId, pending);
}
pending.timer = setTimeout(() => {
const entry = this.pendingNotifications.get(userId);
this.pendingNotifications.delete(userId);
if (entry) {
this._sendNewOpsNotification(userId, entry.excludeClientIds, entry.latestSeq);
}
}, WebSocketConnectionService.NOTIFY_DEBOUNCE_MS);
}
private _sendNewOpsNotification(
userId: number,
excludeClientIds: Set<string>,
latestSeq: number,
): void {
const userSet = this.connections.get(userId);
if (!userSet) return;
const message = {
type: 'new_ops',
latestSeq,
timestamp: Date.now(),
};
let notified = 0;
for (const client of userSet) {
if (!excludeClientIds.has(client.clientId)) {
if (this._sendMessage(client.ws, message)) {
notified++;
}
}
}
if (notified > 0) {
Logger.debug(
`[ws:user:${userId}] Notified ${notified} client(s) about new ops (seq=${latestSeq})`,
);
}
}
startHeartbeat(): void {
if (this.heartbeatInterval) return;
this.heartbeatInterval = setInterval(() => {
const now = Date.now();
const toRemove: { userId: number; client: ConnectedClient }[] = [];
for (const [userId, userSet] of this.connections) {
for (const client of userSet) {
// Check if client responded to last ping
if (
now - client.lastPong >
WebSocketConnectionService.PING_INTERVAL_MS +
WebSocketConnectionService.PONG_TIMEOUT_MS
) {
Logger.info(
`[ws:user:${userId}:${client.clientId}] Dead connection (no pong), closing`,
);
toRemove.push({ userId, client });
continue;
}
// Send app-level ping
this._sendMessage(client.ws, {
type: 'ping',
timestamp: now,
});
// Also send WebSocket-level ping for proxy keepalive
if (client.ws.readyState === WebSocket.OPEN) {
try {
client.ws.ping();
} catch {
Logger.debug(`[ws:user:${userId}:${client.clientId}] Ping failed`);
}
}
}
}
for (const { userId, client } of toRemove) {
this.removeConnection(userId, client);
}
}, WebSocketConnectionService.PING_INTERVAL_MS);
}
stopHeartbeat(): void {
if (this.heartbeatInterval) {
clearInterval(this.heartbeatInterval);
this.heartbeatInterval = null;
}
// Clear pending notifications
for (const entry of this.pendingNotifications.values()) {
clearTimeout(entry.timer);
}
this.pendingNotifications.clear();
}
/** Close all connections gracefully */
closeAll(): void {
for (const [, userSet] of this.connections) {
for (const client of userSet) {
try {
client.ws.close(1001, 'Server shutting down');
} catch (err) {
Logger.debug(`[ws] Error closing connection during shutdown`, err);
}
}
}
this.connections.clear();
}
/** Get total connection count (for monitoring/health) */
getConnectionCount(): number {
let total = 0;
for (const userSet of this.connections.values()) {
total += userSet.size;
}
return total;
}
private _sendMessage(ws: WebSocket, message: Record<string, unknown>): boolean {
if (ws.readyState !== WebSocket.OPEN) return false;
try {
ws.send(JSON.stringify(message));
return true;
} catch (err) {
Logger.debug(`[ws] Failed to send message`, err);
return false;
}
}
}
// Singleton instance
let wsConnectionService: WebSocketConnectionService | null = null;
export const getWsConnectionService = (): WebSocketConnectionService => {
if (!wsConnectionService) {
wsConnectionService = new WebSocketConnectionService();
}
return wsConnectionService;
};
export const resetWsConnectionService = (): void => {
if (wsConnectionService) {
wsConnectionService.stopHeartbeat();
wsConnectionService.closeAll();
}
wsConnectionService = null;
};

View file

@ -0,0 +1,2 @@
export const CLIENT_ID_REGEX = /^[a-zA-Z0-9_-]+$/;
export const MAX_CLIENT_ID_LENGTH = 255;

View file

@ -5,6 +5,7 @@ import { promisify } from 'util';
import { uuidv7 } from 'uuidv7'; import { uuidv7 } from 'uuidv7';
import { authenticate, getAuthUser } from '../middleware'; import { authenticate, getAuthUser } from '../middleware';
import { getSyncService } from './sync.service'; import { getSyncService } from './sync.service';
import { getWsConnectionService } from './services/websocket-connection.service';
import { Logger } from '../logger'; import { Logger } from '../logger';
import { prisma } from '../db'; import { prisma } from '../db';
import { import {
@ -34,8 +35,7 @@ const createValidationErrorResponse = (
}; };
// Validation constants // Validation constants
const CLIENT_ID_REGEX = /^[a-zA-Z0-9_-]+$/; import { CLIENT_ID_REGEX, MAX_CLIENT_ID_LENGTH } from './sync.const';
const MAX_CLIENT_ID_LENGTH = 255;
// Two-stage protection against zip bombs: // Two-stage protection against zip bombs:
// 1. Pre-check: Reject compressed data > limit (typical ratio ~10:1) // 1. Pre-check: Reject compressed data > limit (typical ratio ~10:1)
@ -477,6 +477,11 @@ export const syncRoutes = async (fastify: FastifyInstance): Promise<void> => {
...(hasMorePiggyback ? { hasMorePiggyback: true } : {}), ...(hasMorePiggyback ? { hasMorePiggyback: true } : {}),
}; };
// Notify other connected clients about new ops (fire-and-forget)
if (accepted > 0) {
getWsConnectionService().notifyNewOps(userId, clientId, latestSeq);
}
return reply.send(response); return reply.send(response);
} catch (err) { } catch (err) {
Logger.error(`Upload ops error: ${errorMessage(err)}`); Logger.error(`Upload ops error: ${errorMessage(err)}`);
@ -763,6 +768,11 @@ export const syncRoutes = async (fastify: FastifyInstance): Promise<void> => {
Logger.info(`Snapshot uploaded for user ${userId}, reason: ${reason}`); Logger.info(`Snapshot uploaded for user ${userId}, reason: ${reason}`);
// Notify other connected clients about snapshot upload (fire-and-forget)
if (result.accepted && result.serverSeq !== undefined) {
getWsConnectionService().notifyNewOps(userId, clientId, result.serverSeq);
}
return reply.send({ return reply.send({
accepted: result.accepted, accepted: result.accepted,
serverSeq: result.serverSeq, serverSeq: result.serverSeq,

View file

@ -0,0 +1,68 @@
import { FastifyInstance, FastifyRequest } from 'fastify';
import { verifyToken } from '../auth';
import { getWsConnectionService } from './services/websocket-connection.service';
import { Logger } from '../logger';
import { CLIENT_ID_REGEX, MAX_CLIENT_ID_LENGTH } from './sync.const';
export const wsRoutes = async (fastify: FastifyInstance): Promise<void> => {
fastify.get(
'/ws',
{
websocket: true,
config: {
rateLimit: {
max: 10,
timeWindow: '1 minute',
},
},
},
async (
socket,
req: FastifyRequest<{
Querystring: { token?: string; clientId?: string };
}>,
) => {
try {
const { token, clientId } = req.query as {
token?: string;
clientId?: string;
};
// Validate token
if (!token) {
Logger.warn('[ws] Connection rejected: missing token');
socket.close(4001, 'Missing token');
return;
}
// Validate clientId
if (
!clientId ||
!CLIENT_ID_REGEX.test(clientId) ||
clientId.length > MAX_CLIENT_ID_LENGTH
) {
Logger.warn('[ws] Connection rejected: invalid clientId');
socket.close(4001, 'Invalid clientId');
return;
}
const result = await verifyToken(token);
if (!result.valid) {
Logger.warn(`[ws] Connection rejected: ${result.reason}`);
socket.close(4003, 'Invalid token');
return;
}
const wsService = getWsConnectionService();
wsService.addConnection(result.userId, clientId, socket);
} catch (err) {
Logger.error('[ws] Unexpected error in WebSocket handler:', err);
try {
socket.close(1011, 'Internal error');
} catch (closeErr) {
Logger.debug('[ws] Failed to close socket after error', closeErr);
}
}
},
);
};

View file

@ -0,0 +1,281 @@
import { describe, it, expect, beforeEach, vi, afterEach } from 'vitest';
import { WebSocketConnectionService } from '../src/sync/services/websocket-connection.service';
vi.mock('../src/logger', () => ({
Logger: {
info: vi.fn(),
warn: vi.fn(),
debug: vi.fn(),
error: vi.fn(),
},
}));
const WS_OPEN = 1;
const WS_CLOSED = 3;
interface MockWs {
readyState: number;
send: ReturnType<typeof vi.fn>;
close: ReturnType<typeof vi.fn>;
ping: ReturnType<typeof vi.fn>;
on: ReturnType<typeof vi.fn>;
_handlers: Map<string, (...args: unknown[]) => void>;
_emitPong: () => void;
_emitClose: () => void;
_emitMessage: (data: string) => void;
_emitError: (err: Error) => void;
}
function createMockWs(): MockWs {
const handlers = new Map<string, (...args: unknown[]) => void>();
const mock: MockWs = {
readyState: WS_OPEN,
send: vi.fn(),
close: vi.fn(),
ping: vi.fn(),
on: vi.fn((event: string, handler: (...args: unknown[]) => void) => {
handlers.set(event, handler);
}),
_handlers: handlers,
_emitPong() {
handlers.get('pong')?.();
},
_emitClose() {
handlers.get('close')?.();
},
_emitMessage(data: string) {
handlers.get('message')?.(Buffer.from(data));
},
_emitError(err: Error) {
handlers.get('error')?.(err);
},
};
return mock;
}
function parseSendCalls(mockWs: MockWs): Record<string, unknown>[] {
return mockWs.send.mock.calls.map((c: unknown[]) => JSON.parse(c[0] as string));
}
describe('WebSocketConnectionService', () => {
let service: WebSocketConnectionService;
beforeEach(() => {
vi.useFakeTimers();
service = new WebSocketConnectionService();
});
afterEach(() => {
service.stopHeartbeat();
service.closeAll();
vi.useRealTimers();
});
describe('addConnection', () => {
it('should track a connection and send "connected" message', () => {
const ws = createMockWs();
service.addConnection(1, 'client-a', ws as any);
expect(service.getConnectionCount()).toBe(1);
const messages = parseSendCalls(ws);
expect(messages).toContainEqual(expect.objectContaining({ type: 'connected' }));
});
it('should enforce max 10 connections per user', () => {
const sockets: MockWs[] = [];
for (let i = 0; i < 10; i++) {
const ws = createMockWs();
sockets.push(ws);
service.addConnection(1, `client-${i}`, ws as any);
}
expect(service.getConnectionCount()).toBe(10);
const eleventhWs = createMockWs();
service.addConnection(1, 'client-10', eleventhWs as any);
expect(eleventhWs.close).toHaveBeenCalledWith(4008, 'Too many connections');
expect(service.getConnectionCount()).toBe(10);
});
it('should register close handler that removes connection', () => {
const ws = createMockWs();
service.addConnection(1, 'client-a', ws as any);
expect(service.getConnectionCount()).toBe(1);
ws._emitClose();
expect(service.getConnectionCount()).toBe(0);
});
});
describe('removeConnection', () => {
it('should clean up empty user sets', () => {
const ws = createMockWs();
service.addConnection(1, 'client-a', ws as any);
expect(service.getConnectionCount()).toBe(1);
ws._emitClose();
expect(service.getConnectionCount()).toBe(0);
});
it('should not call close() on already-closed WebSocket', () => {
const ws = createMockWs();
service.addConnection(1, 'client-a', ws as any);
// Reset call count after addConnection (which may have called send, but not close)
const closeCallsBefore = ws.close.mock.calls.length;
// Mark the socket as already closed
ws.readyState = WS_CLOSED;
// Trigger the close event, which calls removeConnection internally
ws._emitClose();
// removeConnection should NOT have called ws.close since readyState is CLOSED
expect(ws.close.mock.calls.length).toBe(closeCallsBefore);
});
});
describe('notifyNewOps', () => {
it('should send to other clients and exclude sender', () => {
const wsA = createMockWs();
const wsB = createMockWs();
service.addConnection(1, 'A', wsA as any);
service.addConnection(1, 'B', wsB as any);
// Reset send mocks after the "connected" messages
wsA.send.mockClear();
wsB.send.mockClear();
service.notifyNewOps(1, 'A', 5);
vi.advanceTimersByTime(100);
const messagesB = parseSendCalls(wsB);
expect(messagesB).toContainEqual(
expect.objectContaining({ type: 'new_ops', latestSeq: 5 }),
);
const messagesA = parseSendCalls(wsA);
expect(messagesA).not.toContainEqual(expect.objectContaining({ type: 'new_ops' }));
});
it('should debounce rapid calls (latest-seq-wins)', () => {
const wsB = createMockWs();
service.addConnection(1, 'B', wsB as any);
wsB.send.mockClear();
service.notifyNewOps(1, 'A', 3);
service.notifyNewOps(1, 'A', 7);
vi.advanceTimersByTime(100);
const messages = parseSendCalls(wsB);
const newOpsMessages = messages.filter((m) => m.type === 'new_ops');
expect(newOpsMessages).toHaveLength(1);
expect(newOpsMessages[0]).toEqual(
expect.objectContaining({ type: 'new_ops', latestSeq: 7 }),
);
});
it('should accumulate excludeClientIds across debounced calls', () => {
const wsA = createMockWs();
const wsB = createMockWs();
const wsC = createMockWs();
service.addConnection(1, 'A', wsA as any);
service.addConnection(1, 'B', wsB as any);
service.addConnection(1, 'C', wsC as any);
wsA.send.mockClear();
wsB.send.mockClear();
wsC.send.mockClear();
service.notifyNewOps(1, 'A', 3);
service.notifyNewOps(1, 'B', 5);
vi.advanceTimersByTime(100);
// Client A excluded from first call, client B excluded from second
const messagesA = parseSendCalls(wsA);
expect(messagesA).not.toContainEqual(expect.objectContaining({ type: 'new_ops' }));
const messagesB = parseSendCalls(wsB);
expect(messagesB).not.toContainEqual(expect.objectContaining({ type: 'new_ops' }));
// Only client C should receive the notification
const messagesC = parseSendCalls(wsC);
expect(messagesC).toContainEqual(
expect.objectContaining({ type: 'new_ops', latestSeq: 5 }),
);
});
it('should no-op for nonexistent user', () => {
expect(() => service.notifyNewOps(999, 'A', 1)).not.toThrow();
vi.advanceTimersByTime(100);
});
});
describe('startHeartbeat', () => {
it('should send ping at interval', () => {
const ws = createMockWs();
service.addConnection(1, 'client-a', ws as any);
ws.send.mockClear();
service.startHeartbeat();
vi.advanceTimersByTime(30_000);
const messages = parseSendCalls(ws);
expect(messages).toContainEqual(expect.objectContaining({ type: 'ping' }));
expect(ws.ping).toHaveBeenCalled();
});
it('should remove dead connections when no pong response', () => {
const ws = createMockWs();
service.addConnection(1, 'client-a', ws as any);
service.startHeartbeat();
// First heartbeat tick: sends ping, connection is still alive
// (Date.now() - lastPong is 30_000, which is less than 30_000 + 10_000 = 40_000)
vi.advanceTimersByTime(30_000);
expect(service.getConnectionCount()).toBe(1);
// Second heartbeat tick: no pong received, so Date.now() - lastPong = 60_000 > 40_000
vi.advanceTimersByTime(30_000);
expect(service.getConnectionCount()).toBe(0);
});
it('should be idempotent', () => {
service.startHeartbeat();
service.startHeartbeat();
const ws = createMockWs();
service.addConnection(1, 'client-a', ws as any);
ws.send.mockClear();
vi.advanceTimersByTime(30_000);
const pingMessages = parseSendCalls(ws).filter((m) => m.type === 'ping');
expect(pingMessages).toHaveLength(1);
});
});
describe('closeAll', () => {
it('should close all connections with code 1001', () => {
const ws1 = createMockWs();
const ws2 = createMockWs();
service.addConnection(1, 'client-a', ws1 as any);
service.addConnection(2, 'client-b', ws2 as any);
expect(service.getConnectionCount()).toBe(2);
service.closeAll();
expect(ws1.close).toHaveBeenCalledWith(1001, 'Server shutting down');
expect(ws2.close).toHaveBeenCalledWith(1001, 'Server shutting down');
expect(service.getConnectionCount()).toBe(0);
});
});
});

View file

@ -0,0 +1,249 @@
import { describe, it, expect, beforeEach, vi } from 'vitest';
import { CLIENT_ID_REGEX, MAX_CLIENT_ID_LENGTH } from '../src/sync/sync.const';
/**
* Tests the WebSocket route validation logic from websocket.routes.ts.
*
* The route handler performs three sequential validations before accepting a connection:
* 1. Token must be present
* 2. ClientId must match CLIENT_ID_REGEX and be within MAX_CLIENT_ID_LENGTH
* 3. Token must pass verifyToken() check
*
* Since Fastify's inject() does not support WebSocket upgrades, we test the
* validation logic directly the regex, length check, and handler flow
* rather than spinning up a real server.
*/
vi.mock('../src/logger', () => ({
Logger: {
info: vi.fn(),
warn: vi.fn(),
debug: vi.fn(),
error: vi.fn(),
},
}));
const mockAddConnection = vi.fn();
vi.mock('../src/sync/services/websocket-connection.service', () => ({
getWsConnectionService: () => ({
addConnection: mockAddConnection,
}),
resetWsConnectionService: vi.fn(),
}));
// Import the already-mocked verifyToken (from tests/setup.ts)
const { verifyToken } = await import('../src/auth');
/**
* Simulates the WebSocket route handler logic from websocket.routes.ts.
* This mirrors the exact validation flow without needing @fastify/websocket.
*/
async function simulateWsHandler(
query: { token?: string; clientId?: string },
socket: { close: ReturnType<typeof vi.fn> },
): Promise<'accepted' | 'rejected'> {
// Dynamic import to pick up the vi.mock above
const { getWsConnectionService } = await import(
'../src/sync/services/websocket-connection.service'
);
try {
const { token, clientId } = query;
if (!token) {
socket.close(4001, 'Missing token');
return 'rejected';
}
if (
!clientId ||
!CLIENT_ID_REGEX.test(clientId) ||
clientId.length > MAX_CLIENT_ID_LENGTH
) {
socket.close(4001, 'Invalid clientId');
return 'rejected';
}
const result = await verifyToken(token);
if (!result.valid) {
socket.close(4003, 'Invalid token');
return 'rejected';
}
const wsService = getWsConnectionService();
wsService.addConnection(result.userId, clientId, socket as any);
return 'accepted';
} catch {
try {
socket.close(1011, 'Internal error');
} catch {
// ignore close error
}
return 'rejected';
}
}
describe('WebSocket Route Validation', () => {
let mockSocket: { close: ReturnType<typeof vi.fn> };
beforeEach(() => {
mockSocket = { close: vi.fn() };
mockAddConnection.mockReset();
vi.mocked(verifyToken).mockResolvedValue({
valid: true,
userId: 1,
email: 'test@test.com',
});
});
describe('CLIENT_ID_REGEX', () => {
it('should accept alphanumeric characters', () => {
expect(CLIENT_ID_REGEX.test('abc123')).toBe(true);
});
it('should accept underscores and hyphens', () => {
expect(CLIENT_ID_REGEX.test('client_ID-123')).toBe(true);
});
it('should reject special characters', () => {
expect(CLIENT_ID_REGEX.test('invalid!@#')).toBe(false);
});
it('should reject spaces', () => {
expect(CLIENT_ID_REGEX.test('has space')).toBe(false);
});
it('should reject empty string', () => {
expect(CLIENT_ID_REGEX.test('')).toBe(false);
});
it('should reject dots', () => {
expect(CLIENT_ID_REGEX.test('client.id')).toBe(false);
});
});
describe('MAX_CLIENT_ID_LENGTH', () => {
it('should be 255', () => {
expect(MAX_CLIENT_ID_LENGTH).toBe(255);
});
});
describe('handler validation flow', () => {
it('should reject when token is missing', async () => {
const result = await simulateWsHandler(
{ clientId: 'valid_client' },
mockSocket,
);
expect(result).toBe('rejected');
expect(mockSocket.close).toHaveBeenCalledWith(4001, 'Missing token');
});
it('should reject when token is empty string', async () => {
const result = await simulateWsHandler(
{ token: '', clientId: 'valid_client' },
mockSocket,
);
expect(result).toBe('rejected');
expect(mockSocket.close).toHaveBeenCalledWith(4001, 'Missing token');
});
it('should reject when clientId is missing', async () => {
const result = await simulateWsHandler({ token: 'some-token' }, mockSocket);
expect(result).toBe('rejected');
expect(mockSocket.close).toHaveBeenCalledWith(4001, 'Invalid clientId');
});
it('should reject when clientId has invalid characters', async () => {
const result = await simulateWsHandler(
{ token: 'some-token', clientId: 'bad!!!id' },
mockSocket,
);
expect(result).toBe('rejected');
expect(mockSocket.close).toHaveBeenCalledWith(4001, 'Invalid clientId');
});
it('should reject when clientId exceeds max length', async () => {
const result = await simulateWsHandler(
{ token: 'some-token', clientId: 'a'.repeat(256) },
mockSocket,
);
expect(result).toBe('rejected');
expect(mockSocket.close).toHaveBeenCalledWith(4001, 'Invalid clientId');
});
it('should accept clientId at exactly max length', async () => {
const result = await simulateWsHandler(
{ token: 'good-token', clientId: 'a'.repeat(255) },
mockSocket,
);
expect(result).toBe('accepted');
expect(mockSocket.close).not.toHaveBeenCalled();
expect(mockAddConnection).toHaveBeenCalledWith(1, 'a'.repeat(255), mockSocket);
});
it('should reject when verifyToken returns invalid', async () => {
vi.mocked(verifyToken).mockResolvedValue({
valid: false,
reason: 'token expired',
});
const result = await simulateWsHandler(
{ token: 'expired-token', clientId: 'client_1' },
mockSocket,
);
expect(result).toBe('rejected');
expect(mockSocket.close).toHaveBeenCalledWith(4003, 'Invalid token');
expect(verifyToken).toHaveBeenCalledWith('expired-token');
});
it('should accept valid connection and call addConnection', async () => {
const result = await simulateWsHandler(
{ token: 'good-token', clientId: 'client_1' },
mockSocket,
);
expect(result).toBe('accepted');
expect(verifyToken).toHaveBeenCalledWith('good-token');
expect(mockAddConnection).toHaveBeenCalledWith(1, 'client_1', mockSocket);
expect(mockSocket.close).not.toHaveBeenCalled();
});
it('should close with 1011 when verifyToken throws', async () => {
vi.mocked(verifyToken).mockRejectedValue(new Error('database down'));
const result = await simulateWsHandler(
{ token: 'some-token', clientId: 'client_1' },
mockSocket,
);
expect(result).toBe('rejected');
expect(mockSocket.close).toHaveBeenCalledWith(1011, 'Internal error');
});
it('should validate token before clientId format', async () => {
const result = await simulateWsHandler({ clientId: 'bad!!!id' }, mockSocket);
expect(result).toBe('rejected');
expect(mockSocket.close).toHaveBeenCalledWith(4001, 'Missing token');
expect(verifyToken).not.toHaveBeenCalled();
});
it('should validate clientId before calling verifyToken', async () => {
const result = await simulateWsHandler(
{ token: 'some-token', clientId: 'bad!!!id' },
mockSocket,
);
expect(result).toBe('rejected');
expect(verifyToken).not.toHaveBeenCalled();
});
});
});

View file

@ -1,6 +1,6 @@
import { signal } from '@angular/core'; import { signal } from '@angular/core';
import { TestBed } from '@angular/core/testing'; import { TestBed } from '@angular/core/testing';
import { BehaviorSubject, of } from 'rxjs'; import { BehaviorSubject, firstValueFrom, of } from 'rxjs';
import { SyncWrapperService } from './sync-wrapper.service'; import { SyncWrapperService } from './sync-wrapper.service';
import { SyncProviderManager } from '../../op-log/sync-providers/provider-manager.service'; import { SyncProviderManager } from '../../op-log/sync-providers/provider-manager.service';
import { OperationLogSyncService } from '../../op-log/sync/operation-log-sync.service'; import { OperationLogSyncService } from '../../op-log/sync/operation-log-sync.service';
@ -15,6 +15,8 @@ import { ReminderService } from '../../features/reminder/reminder.service';
import { DataInitService } from '../../core/data-init/data-init.service'; import { DataInitService } from '../../core/data-init/data-init.service';
import { UserInputWaitStateService } from './user-input-wait-state.service'; import { UserInputWaitStateService } from './user-input-wait-state.service';
import { SuperSyncStatusService } from '../../op-log/sync/super-sync-status.service'; import { SuperSyncStatusService } from '../../op-log/sync/super-sync-status.service';
import { SuperSyncWebSocketService } from '../../op-log/sync/super-sync-websocket.service';
import { WsTriggeredDownloadService } from '../../op-log/sync/ws-triggered-download.service';
import { import {
AuthFailSPError, AuthFailSPError,
MissingCredentialsSPError, MissingCredentialsSPError,
@ -44,14 +46,23 @@ describe('SyncWrapperService', () => {
let mockReminderService: jasmine.SpyObj<ReminderService>; let mockReminderService: jasmine.SpyObj<ReminderService>;
let mockUserInputWaitState: jasmine.SpyObj<UserInputWaitStateService>; let mockUserInputWaitState: jasmine.SpyObj<UserInputWaitStateService>;
let mockSuperSyncStatusService: jasmine.SpyObj<SuperSyncStatusService>; let mockSuperSyncStatusService: jasmine.SpyObj<SuperSyncStatusService>;
let mockSuperSyncWsService: jasmine.SpyObj<SuperSyncWebSocketService> & {
isConnected: ReturnType<typeof signal<boolean>>;
};
let mockWsDownloadService: jasmine.SpyObj<WsTriggeredDownloadService>;
let configSubject: BehaviorSubject<any>; let configSubject: BehaviorSubject<any>;
let mockSyncCapableProvider: any; let mockSyncCapableProvider: any;
const createMockSyncConfig = (provider: SyncProviderId | null): { sync: any } => ({ const createMockSyncConfig = (
provider: SyncProviderId | null,
overrides: Record<string, unknown> = {},
): { sync: any } => ({
sync: { sync: {
syncProvider: provider, syncProvider: provider,
syncInterval: 60000, syncInterval: 60000,
isManualSyncOnly: false,
...overrides,
}, },
}); });
@ -163,6 +174,19 @@ describe('SyncWrapperService', () => {
}, },
); );
mockSuperSyncWsService = Object.assign(
jasmine.createSpyObj('SuperSyncWebSocketService', ['connect', 'disconnect']),
{
isConnected: signal(false),
},
);
mockSuperSyncWsService.connect.and.returnValue(Promise.resolve());
mockWsDownloadService = jasmine.createSpyObj('WsTriggeredDownloadService', [
'start',
'stop',
]);
TestBed.configureTestingModule({ TestBed.configureTestingModule({
providers: [ providers: [
SyncWrapperService, SyncWrapperService,
@ -179,6 +203,8 @@ describe('SyncWrapperService', () => {
{ provide: ReminderService, useValue: mockReminderService }, { provide: ReminderService, useValue: mockReminderService },
{ provide: UserInputWaitStateService, useValue: mockUserInputWaitState }, { provide: UserInputWaitStateService, useValue: mockUserInputWaitState },
{ provide: SuperSyncStatusService, useValue: mockSuperSyncStatusService }, { provide: SuperSyncStatusService, useValue: mockSuperSyncStatusService },
{ provide: SuperSyncWebSocketService, useValue: mockSuperSyncWsService },
{ provide: WsTriggeredDownloadService, useValue: mockWsDownloadService },
], ],
}); });
@ -233,6 +259,109 @@ describe('SyncWrapperService', () => {
}); });
}); });
describe('syncInterval$', () => {
it('should use 1 minute for SuperSync when websocket is disconnected', async () => {
expect(await firstValueFrom(service.syncInterval$)).toBe(60000);
});
it('should use 5 minutes for SuperSync when websocket is connected', async () => {
mockSuperSyncWsService.isConnected.set(true);
configSubject.next(createMockSyncConfig(SyncProviderId.SuperSync));
expect(await firstValueFrom(service.syncInterval$)).toBe(300000);
});
it('should return 0 for manual sync only', async () => {
configSubject.next(
createMockSyncConfig(SyncProviderId.SuperSync, { isManualSyncOnly: true }),
);
expect(await firstValueFrom(service.syncInterval$)).toBe(0);
});
it('should use configured interval for non-SuperSync providers', async () => {
configSubject.next(
createMockSyncConfig(SyncProviderId.WebDAV, { syncInterval: 120000 }),
);
expect(await firstValueFrom(service.syncInterval$)).toBe(120000);
});
});
describe('websocket integration', () => {
it('should disconnect websocket when provider changes away from SuperSync', async () => {
configSubject.next(createMockSyncConfig(SyncProviderId.WebDAV));
await Promise.resolve();
expect(mockWsDownloadService.stop).toHaveBeenCalled();
expect(mockSuperSyncWsService.disconnect).toHaveBeenCalled();
});
it('should connect websocket after successful SuperSync sync', async () => {
const mockProvider = {
getWebSocketParams: jasmine.createSpy().and.returnValue(
Promise.resolve({
baseUrl: 'https://sync.example.com',
accessToken: 'token-123',
}),
),
};
mockProviderManager.getProviderById.and.returnValue(
Promise.resolve(mockProvider as any),
);
await service.sync();
// Flush microtasks: connectWebSocket() is fire-and-forget (not awaited in _sync).
// Two flushes needed: one for getProviderById, one for getWebSocketParams + connect.
await Promise.resolve();
await Promise.resolve();
expect(mockProviderManager.getProviderById).toHaveBeenCalledWith(
SyncProviderId.SuperSync,
);
expect(mockProvider.getWebSocketParams).toHaveBeenCalled();
expect(mockSuperSyncWsService.connect).toHaveBeenCalledWith(
'https://sync.example.com',
'token-123',
);
expect(mockWsDownloadService.start).toHaveBeenCalled();
});
it('should no-op connectWebSocket when provider is not SuperSync', async () => {
configSubject.next(createMockSyncConfig(SyncProviderId.WebDAV));
await service.connectWebSocket();
await Promise.resolve();
expect(mockProviderManager.getProviderById).not.toHaveBeenCalled();
});
it('should no-op connectWebSocket when getWebSocketParams returns null', async () => {
const mockProvider = {
getWebSocketParams: jasmine.createSpy().and.returnValue(Promise.resolve(null)),
};
mockProviderManager.getProviderById.and.returnValue(
Promise.resolve(mockProvider as any),
);
await service.connectWebSocket();
await Promise.resolve();
await Promise.resolve();
expect(mockProvider.getWebSocketParams).toHaveBeenCalled();
expect(mockSuperSyncWsService.connect).not.toHaveBeenCalled();
});
it('should skip WS connection after sync if already connected', async () => {
mockSuperSyncWsService.isConnected.set(true);
await service.sync();
await Promise.resolve();
await Promise.resolve();
expect(mockSuperSyncWsService.connect).not.toHaveBeenCalled();
});
});
describe('_sync() - Provider handling', () => { describe('_sync() - Provider handling', () => {
it('should call _syncVectorClockToPfapi for WebDAV provider', async () => { it('should call _syncVectorClockToPfapi for WebDAV provider', async () => {
configSubject.next(createMockSyncConfig(SyncProviderId.WebDAV)); configSubject.next(createMockSyncConfig(SyncProviderId.WebDAV));

View file

@ -1,5 +1,5 @@
import { inject, Injectable } from '@angular/core'; import { DestroyRef, inject, Injectable } from '@angular/core';
import { BehaviorSubject, firstValueFrom, Observable, of } from 'rxjs'; import { BehaviorSubject, combineLatest, firstValueFrom, Observable, of } from 'rxjs';
import { GlobalConfigService } from '../../features/config/global-config.service'; import { GlobalConfigService } from '../../features/config/global-config.service';
import { import {
distinctUntilChanged, distinctUntilChanged,
@ -8,10 +8,9 @@ import {
map, map,
shareReplay, shareReplay,
switchMap, switchMap,
take,
timeout, timeout,
} from 'rxjs/operators'; } from 'rxjs/operators';
import { toObservable } from '@angular/core/rxjs-interop'; import { takeUntilDestroyed, toObservable } from '@angular/core/rxjs-interop';
import { import {
SyncAlreadyInProgressError, SyncAlreadyInProgressError,
LockAcquisitionTimeoutError, LockAcquisitionTimeoutError,
@ -64,10 +63,13 @@ import { alertDialog, confirmDialog } from '../../util/native-dialogs';
import { UserInputWaitStateService } from './user-input-wait-state.service'; import { UserInputWaitStateService } from './user-input-wait-state.service';
import { SYNC_WAIT_TIMEOUT_MS } from './sync.const'; import { SYNC_WAIT_TIMEOUT_MS } from './sync.const';
import { SuperSyncStatusService } from '../../op-log/sync/super-sync-status.service'; import { SuperSyncStatusService } from '../../op-log/sync/super-sync-status.service';
import { SuperSyncWebSocketService } from '../../op-log/sync/super-sync-websocket.service';
import { WsTriggeredDownloadService } from '../../op-log/sync/ws-triggered-download.service';
import { IS_ELECTRON } from '../../app.constants'; import { IS_ELECTRON } from '../../app.constants';
import { OperationLogStoreService } from '../../op-log/persistence/operation-log-store.service'; import { OperationLogStoreService } from '../../op-log/persistence/operation-log-store.service';
import { OperationLogSyncService } from '../../op-log/sync/operation-log-sync.service'; import { OperationLogSyncService } from '../../op-log/sync/operation-log-sync.service';
import { WrappedProviderService } from '../../op-log/sync-providers/wrapped-provider.service'; import { WrappedProviderService } from '../../op-log/sync-providers/wrapped-provider.service';
import { SuperSyncProvider } from '../../op-log/sync-providers/super-sync/super-sync';
@Injectable({ @Injectable({
providedIn: 'root', providedIn: 'root',
@ -83,6 +85,8 @@ export class SyncWrapperService {
private _reminderService = inject(ReminderService); private _reminderService = inject(ReminderService);
private _userInputWaitState = inject(UserInputWaitStateService); private _userInputWaitState = inject(UserInputWaitStateService);
private _superSyncStatusService = inject(SuperSyncStatusService); private _superSyncStatusService = inject(SuperSyncStatusService);
private _superSyncWsService = inject(SuperSyncWebSocketService);
private _wsDownloadService = inject(WsTriggeredDownloadService);
private _opLogStore = inject(OperationLogStoreService); private _opLogStore = inject(OperationLogStoreService);
private _opLogSyncService = inject(OperationLogSyncService); private _opLogSyncService = inject(OperationLogSyncService);
private _wrappedProvider = inject(WrappedProviderService); private _wrappedProvider = inject(WrappedProviderService);
@ -96,13 +100,36 @@ export class SyncWrapperService {
map((cfg) => toSyncProviderId(cfg.syncProvider)), map((cfg) => toSyncProviderId(cfg.syncProvider)),
); );
// SuperSync always uses 1 minute interval; other providers use configured value private _destroyRef = inject(DestroyRef);
// Return 0 when manual sync only is enabled to disable automatic triggers
syncInterval$: Observable<number> = this.syncCfg$.pipe( // Disconnect WebSocket when sync provider changes away from SuperSync or sync is disabled
map((cfg) => { private _wsProviderCleanup = this.syncProviderId$
.pipe(distinctUntilChanged(), takeUntilDestroyed(this._destroyRef))
.subscribe((providerId) => {
if (providerId !== SyncProviderId.SuperSync) {
this.disconnectWebSocket();
}
});
/**
* Sync interval in milliseconds.
* - When WebSocket is connected: 5 minutes (health check only)
* - When WebSocket is disconnected: 1 minute for SuperSync
* - Other providers: user-configured value
* - Return 0 when manual sync only is enabled to disable automatic triggers
*/
syncInterval$: Observable<number> = combineLatest([
this.syncCfg$,
toObservable(this._superSyncWsService.isConnected),
]).pipe(
map(([cfg, wsConnected]) => {
if (cfg.isManualSyncOnly) return 0; if (cfg.isManualSyncOnly) return 0;
return cfg.syncProvider === SyncProviderId.SuperSync ? 60000 : cfg.syncInterval; if (cfg.syncProvider === SyncProviderId.SuperSync) {
return wsConnected ? 300_000 : 60_000;
}
return cfg.syncInterval;
}), }),
distinctUntilChanged(),
); );
isEnabledAndReady$: Observable<boolean> = this._providerManager.isProviderReady$; isEnabledAndReady$: Observable<boolean> = this._providerManager.isProviderReady$;
@ -279,8 +306,49 @@ export class SyncWrapperService {
} }
} }
/**
* Connects the WebSocket for real-time sync notifications.
* Called after a successful SuperSync sync cycle.
*/
async connectWebSocket(): Promise<void> {
const providerId = await firstValueFrom(this.syncProviderId$);
if (providerId !== SyncProviderId.SuperSync) {
return;
}
const provider = await this._providerManager.getProviderById(
SyncProviderId.SuperSync,
);
if (!provider) {
SyncLog.warn(
'SyncWrapperService: No SuperSync provider found for WebSocket connection',
);
return;
}
const superSyncProvider = provider as unknown as SuperSyncProvider;
const wsParams = await superSyncProvider.getWebSocketParams();
if (!wsParams) {
SyncLog.warn(
'SyncWrapperService: No WebSocket params available from SuperSync provider',
);
return;
}
await this._superSyncWsService.connect(wsParams.baseUrl, wsParams.accessToken);
this._wsDownloadService.start();
}
/**
* Disconnects the WebSocket and stops WS-triggered downloads.
*/
disconnectWebSocket(): void {
this._wsDownloadService.stop();
this._superSyncWsService.disconnect();
}
private async _sync(): Promise<SyncStatus | 'HANDLED_ERROR'> { private async _sync(): Promise<SyncStatus | 'HANDLED_ERROR'> {
const providerId = await this.syncProviderId$.pipe(take(1)).toPromise(); const providerId = await firstValueFrom(this.syncProviderId$);
if (!providerId) { if (!providerId) {
throw new Error('No Sync Provider for sync()'); throw new Error('No Sync Provider for sync()');
} }
@ -414,6 +482,20 @@ export class SyncWrapperService {
// Mark as in-sync for all providers after successful sync // Mark as in-sync for all providers after successful sync
this._providerManager.setSyncStatus('IN_SYNC'); this._providerManager.setSyncStatus('IN_SYNC');
SyncLog.log('SyncWrapperService: Sync complete, status=IN_SYNC'); SyncLog.log('SyncWrapperService: Sync complete, status=IN_SYNC');
// Connect WebSocket after first successful SuperSync sync (fire-and-forget)
if (
providerId === SyncProviderId.SuperSync &&
!this._superSyncWsService.isConnected()
) {
this.connectWebSocket().catch((err) => {
SyncLog.warn(
'SyncWrapperService: WebSocket connection failed, will retry on next sync',
err,
);
});
}
return SyncStatus.InSync; return SyncStatus.InSync;
} catch (error) { } catch (error) {
SyncLog.err(error); SyncLog.err(error);

View file

@ -225,6 +225,41 @@ describe('SuperSyncProvider', () => {
}); });
}); });
describe('getWebSocketParams', () => {
it('should return null when access token is missing', async () => {
mockPrivateCfgStore.load.and.returnValue(
Promise.resolve({ baseUrl: 'https://sync.example.com' } as SuperSyncPrivateCfg),
);
expect(await provider.getWebSocketParams()).toBeNull();
});
it('should return sanitized params using the configured base URL', async () => {
mockPrivateCfgStore.load.and.returnValue(
Promise.resolve({
baseUrl: 'https://sync.example.com/',
accessToken: 'token\u200b123',
} as SuperSyncPrivateCfg),
);
await expectAsync(provider.getWebSocketParams()).toBeResolvedTo({
baseUrl: 'https://sync.example.com',
accessToken: 'token123',
});
});
it('should fall back to the default base URL when none is configured', async () => {
mockPrivateCfgStore.load.and.returnValue(
Promise.resolve({ accessToken: 'test-access-token' } as SuperSyncPrivateCfg),
);
await expectAsync(provider.getWebSocketParams()).toBeResolvedTo({
baseUrl: SUPER_SYNC_DEFAULT_BASE_URL,
accessToken: 'test-access-token',
});
});
});
describe('uploadOps', () => { describe('uploadOps', () => {
it('should upload operations successfully', async () => { it('should upload operations successfully', async () => {
mockPrivateCfgStore.load.and.returnValue(Promise.resolve(testConfig)); mockPrivateCfgStore.load.and.returnValue(Promise.resolve(testConfig));

View file

@ -311,6 +311,26 @@ export class SuperSyncProvider
return validateRestoreSnapshotResponse(response); return validateRestoreSnapshotResponse(response);
} }
// === WebSocket Parameters ===
/**
* Returns the base URL and sanitized access token for WebSocket connection.
* Returns null if provider is not configured.
*/
async getWebSocketParams(): Promise<{
baseUrl: string;
accessToken: string;
} | null> {
const cfg = await this.privateCfg.load();
if (!cfg?.accessToken) {
return null;
}
return {
baseUrl: this._resolveBaseUrl(cfg),
accessToken: this._sanitizeToken(cfg.accessToken),
};
}
// === Data Management === // === Data Management ===
async deleteAllData(): Promise<{ success: boolean }> { async deleteAllData(): Promise<{ success: boolean }> {

View file

@ -0,0 +1,277 @@
import { TestBed } from '@angular/core/testing';
import { CLIENT_ID_PROVIDER } from '../util/client-id.provider';
import { SuperSyncWebSocketService } from './super-sync-websocket.service';
class MockWebSocket {
static CONNECTING = 0;
static OPEN = 1;
static CLOSING = 2;
static CLOSED = 3;
static instances: MockWebSocket[] = [];
readyState = MockWebSocket.CONNECTING;
onopen: (() => void) | null = null;
onmessage: ((event: MessageEvent) => void) | null = null;
onclose: ((event: CloseEvent) => void) | null = null;
onerror: ((event: Event) => void) | null = null;
send = jasmine.createSpy('send');
close = jasmine.createSpy('close').and.callFake((_code?: number, _reason?: string) => {
this.readyState = MockWebSocket.CLOSED;
});
constructor(public url: string) {
MockWebSocket.instances.push(this);
}
emitOpen(): void {
this.readyState = MockWebSocket.OPEN;
this.onopen?.();
}
emitMessage(data: unknown): void {
const payload = typeof data === 'string' ? data : JSON.stringify(data);
this.onmessage?.({ data: payload } as MessageEvent);
}
emitClose(code = 1006, reason = ''): void {
this.readyState = MockWebSocket.CLOSED;
this.onclose?.({ code, reason } as CloseEvent);
}
}
describe('SuperSyncWebSocketService', () => {
let service: SuperSyncWebSocketService;
let mockClientIdProvider: { loadClientId: jasmine.Spy };
let originalWebSocket: typeof WebSocket;
beforeEach(() => {
jasmine.clock().install();
MockWebSocket.instances = [];
originalWebSocket = globalThis.WebSocket;
(globalThis as typeof globalThis & { WebSocket: typeof WebSocket }).WebSocket =
MockWebSocket as unknown as typeof WebSocket;
mockClientIdProvider = {
loadClientId: jasmine
.createSpy('loadClientId')
.and.returnValue(Promise.resolve('client_1')),
};
TestBed.configureTestingModule({
providers: [
SuperSyncWebSocketService,
{ provide: CLIENT_ID_PROVIDER, useValue: mockClientIdProvider },
],
});
service = TestBed.inject(SuperSyncWebSocketService);
});
afterEach(() => {
service.disconnect();
jasmine.clock().uninstall();
(globalThis as typeof globalThis & { WebSocket: typeof WebSocket }).WebSocket =
originalWebSocket;
});
it('should create a websocket with the encoded sync URL', async () => {
await service.connect('https://sync.example.com', 'token with +/?');
expect(MockWebSocket.instances).toHaveSize(1);
expect(MockWebSocket.instances[0].url).toBe(
'wss://sync.example.com/api/sync/ws?token=token%20with%20%2B%2F%3F&clientId=client_1',
);
});
it('should emit notifications for new_ops messages', async () => {
const received: number[] = [];
service.newOpsNotification$.subscribe((notification) =>
received.push(notification.latestSeq),
);
await service.connect('http://localhost:1901', 'token');
MockWebSocket.instances[0].emitMessage({ type: 'new_ops', latestSeq: 7 });
expect(received).toEqual([7]);
});
it('should respond to ping messages with pong', async () => {
await service.connect('http://localhost:1901', 'token');
const ws = MockWebSocket.instances[0];
ws.emitOpen();
ws.emitMessage({ type: 'ping' });
expect(ws.send).toHaveBeenCalledWith(JSON.stringify({ type: 'pong' }));
});
it('should reconnect after an unexpected close', async () => {
spyOn(Math, 'random').and.returnValue(0.5);
await service.connect('http://localhost:1901', 'token');
MockWebSocket.instances[0].emitClose(1006, 'network error');
jasmine.clock().tick(1000);
await Promise.resolve();
expect(MockWebSocket.instances).toHaveSize(2);
expect(MockWebSocket.instances[1].url).toContain('/api/sync/ws?token=token');
});
it('should not reconnect after auth failure', async () => {
await service.connect('http://localhost:1901', 'token');
MockWebSocket.instances[0].emitClose(4003, 'Invalid token');
jasmine.clock().tick(60000);
await Promise.resolve();
expect(MockWebSocket.instances).toHaveSize(1);
});
it('should close the connection after heartbeat timeout', async () => {
await service.connect('http://localhost:1901', 'token');
const ws = MockWebSocket.instances[0];
ws.emitOpen();
jasmine.clock().tick(45000);
expect(ws.close).toHaveBeenCalledWith(4000, 'Heartbeat timeout');
});
it('should skip connecting when no client id is available', async () => {
mockClientIdProvider.loadClientId.and.returnValue(Promise.resolve(null));
await service.connect('http://localhost:1901', 'token');
expect(MockWebSocket.instances).toHaveSize(0);
});
it('should clear state on disconnect and be safe to call twice', async () => {
await service.connect('http://localhost:1901', 'token');
const ws = MockWebSocket.instances[0];
ws.emitOpen();
expect(service.isConnected()).toBe(true);
service.disconnect();
expect(service.isConnected()).toBe(false);
expect(ws.close).toHaveBeenCalled();
// Calling disconnect a second time should not throw
expect(() => service.disconnect()).not.toThrow();
});
it('should stop reconnecting after 50 attempts', async () => {
spyOn(Math, 'random').and.returnValue(0.5);
await service.connect('http://localhost:1901', 'token');
for (let i = 0; i < 50; i++) {
MockWebSocket.instances[MockWebSocket.instances.length - 1].emitClose(1006, 'drop');
jasmine.clock().tick(60001);
await Promise.resolve();
}
// After the 50th close, schedule reconnect should be a no-op (max reached)
MockWebSocket.instances[MockWebSocket.instances.length - 1].emitClose(1006, 'drop');
jasmine.clock().tick(60001);
await Promise.resolve();
// Original + 50 reconnects = 51, NOT 52
expect(MockWebSocket.instances).toHaveSize(51);
});
it('should use exponential backoff capped at 60 seconds', async () => {
spyOn(Math, 'random').and.returnValue(0.5);
await service.connect('http://localhost:1901', 'token');
// Close #1: baseDelay = 1000 * 2^0 = 1000ms
MockWebSocket.instances[0].emitClose(1006, 'drop');
jasmine.clock().tick(1000);
await Promise.resolve();
expect(MockWebSocket.instances).toHaveSize(2);
// Close #2: baseDelay = 1000 * 2^1 = 2000ms
MockWebSocket.instances[1].emitClose(1006, 'drop');
jasmine.clock().tick(2000);
await Promise.resolve();
expect(MockWebSocket.instances).toHaveSize(3);
// Close #3: baseDelay = 1000 * 2^2 = 4000ms
MockWebSocket.instances[2].emitClose(1006, 'drop');
jasmine.clock().tick(4000);
await Promise.resolve();
expect(MockWebSocket.instances).toHaveSize(4);
// Close #4: baseDelay = 1000 * 2^3 = 8000ms
MockWebSocket.instances[3].emitClose(1006, 'drop');
jasmine.clock().tick(8000);
await Promise.resolve();
expect(MockWebSocket.instances).toHaveSize(5);
// Fast-forward to attempt 7+: cap at 60000ms
// Close #5 and #6 quickly
MockWebSocket.instances[4].emitClose(1006, 'drop');
jasmine.clock().tick(60001);
await Promise.resolve();
expect(MockWebSocket.instances).toHaveSize(6);
MockWebSocket.instances[5].emitClose(1006, 'drop');
jasmine.clock().tick(60001);
await Promise.resolve();
expect(MockWebSocket.instances).toHaveSize(7);
// At attempt 7 (index 6), baseDelay = min(1000 * 2^6, 60000) = 60000
// Verify that 59999ms is NOT enough
MockWebSocket.instances[6].emitClose(1006, 'drop');
jasmine.clock().tick(59999);
await Promise.resolve();
expect(MockWebSocket.instances).toHaveSize(7); // still 7
// But 2ms more (total 60001ms) triggers it
jasmine.clock().tick(2);
await Promise.resolve();
expect(MockWebSocket.instances).toHaveSize(8);
});
it('should handle connected message type without error', async () => {
const received: number[] = [];
service.newOpsNotification$.subscribe((notification) =>
received.push(notification.latestSeq),
);
await service.connect('http://localhost:1901', 'token');
const ws = MockWebSocket.instances[0];
ws.emitOpen();
ws.emitMessage({ type: 'connected' });
expect(received).toEqual([]);
expect(service.isConnected()).toBe(true);
});
it('should convert http:// URL to ws://', async () => {
await service.connect('http://localhost:1901', 'token');
expect(MockWebSocket.instances[0].url).toMatch(/^ws:\/\//);
expect(MockWebSocket.instances[0].url).not.toMatch(/^wss:\/\//);
});
it('should reset heartbeat timer on incoming message', async () => {
await service.connect('http://localhost:1901', 'token');
const ws = MockWebSocket.instances[0];
ws.emitOpen();
// Advance 40s (close to 45s heartbeat timeout but not past it)
jasmine.clock().tick(40000);
ws.emitMessage({ type: 'ping' }); // resets heartbeat timer
// Advance another 40s — only 40s since last message, heartbeat should NOT fire
jasmine.clock().tick(40000);
expect(ws.close).not.toHaveBeenCalled();
// Advance 5001ms more — now 45001ms since last message
jasmine.clock().tick(5001);
expect(ws.close).toHaveBeenCalledWith(4000, 'Heartbeat timeout');
});
});

View file

@ -0,0 +1,249 @@
import { Injectable, inject, signal, OnDestroy } from '@angular/core';
import { Observable, Subject } from 'rxjs';
import { SyncLog } from '../../core/log';
import { CLIENT_ID_PROVIDER } from '../util/client-id.provider';
export interface NewOpsNotification {
latestSeq: number;
}
interface WsMessage {
type: string;
latestSeq?: number;
timestamp?: number;
}
const MIN_RECONNECT_DELAY_MS = 1000;
const MAX_RECONNECT_DELAY_MS = 60_000;
const MAX_RECONNECT_ATTEMPTS = 50;
const JITTER_FACTOR = 0.1;
/** Must be greater than server ping interval (30s) */
const HEARTBEAT_TIMEOUT_MS = 45_000;
/** Close code indicating auth failure - do not reconnect */
const AUTH_FAILURE_CLOSE_CODE = 4003;
@Injectable({
providedIn: 'root',
})
export class SuperSyncWebSocketService implements OnDestroy {
private _clientIdProvider = inject(CLIENT_ID_PROVIDER);
readonly isConnected = signal(false);
private _newOpsNotification$ = new Subject<NewOpsNotification>();
readonly newOpsNotification$: Observable<NewOpsNotification> =
this._newOpsNotification$.asObservable();
private _ws: WebSocket | null = null;
private _reconnectAttempts = 0;
private _reconnectTimer: ReturnType<typeof setTimeout> | null = null;
private _heartbeatTimer: ReturnType<typeof setTimeout> | null = null;
private _isIntentionalClose = false;
private _currentParams: { baseUrl: string; accessToken: string } | null = null;
async connect(baseUrl: string, accessToken: string): Promise<void> {
// Disconnect existing connection first
this.disconnect();
this._isIntentionalClose = false;
this._reconnectAttempts = 0;
this._currentParams = { baseUrl, accessToken };
await this._connect(baseUrl, accessToken);
}
disconnect(): void {
this._isIntentionalClose = true;
this._currentParams = null;
this._clearReconnectTimer();
this._clearHeartbeatTimer();
if (this._ws) {
try {
this._ws.close(1000, 'Client disconnect');
} catch (err) {
SyncLog.warn('SuperSyncWebSocketService: Error closing WebSocket', err);
}
this._ws = null;
}
this.isConnected.set(false);
}
ngOnDestroy(): void {
this.disconnect();
this._newOpsNotification$.complete();
}
private async _connect(baseUrl: string, accessToken: string): Promise<void> {
const clientId = await this._clientIdProvider.loadClientId();
if (!clientId) {
SyncLog.warn('SuperSyncWebSocketService: No clientId available, cannot connect');
return;
}
// Convert http(s) to ws(s)
const wsUrl = baseUrl.replace(/^https:\/\//, 'wss://').replace(/^http:\/\//, 'ws://');
const url = `${wsUrl}/api/sync/ws?token=${encodeURIComponent(accessToken)}&clientId=${encodeURIComponent(clientId)}`;
try {
this._ws = new WebSocket(url);
} catch (err) {
SyncLog.warn('SuperSyncWebSocketService: Failed to create WebSocket', err);
this._scheduleReconnect();
return;
}
this._ws.onopen = (): void => {
SyncLog.log('SuperSyncWebSocketService: Connected');
this._reconnectAttempts = 0;
this.isConnected.set(true);
this._startHeartbeatTimer();
};
this._ws.onmessage = (event: MessageEvent): void => {
this._resetHeartbeatTimer();
let msg: WsMessage;
try {
msg = JSON.parse(event.data as string);
} catch {
SyncLog.warn('SuperSyncWebSocketService: Received non-JSON message, ignoring');
return;
}
try {
this._handleMessage(msg);
} catch (err) {
SyncLog.err('SuperSyncWebSocketService: Error handling message', err);
}
};
this._ws.onclose = (event: CloseEvent): void => {
SyncLog.log(
`SuperSyncWebSocketService: Closed (code=${event.code}, reason=${event.reason})`,
);
this._ws = null;
this.isConnected.set(false);
this._clearHeartbeatTimer();
// Don't reconnect on intentional close or auth failure
if (this._isIntentionalClose) {
return;
}
if (event.code === AUTH_FAILURE_CLOSE_CODE) {
SyncLog.warn(
'SuperSyncWebSocketService: Auth failure, not reconnecting. Waiting for re-auth.',
);
return;
}
this._scheduleReconnect();
};
this._ws.onerror = (): void => {
// Error is followed by close event, so reconnection is handled there
SyncLog.warn('SuperSyncWebSocketService: WebSocket error');
};
}
private _handleMessage(msg: WsMessage): void {
switch (msg.type) {
case 'new_ops':
if (msg.latestSeq !== undefined) {
this._newOpsNotification$.next({ latestSeq: msg.latestSeq });
}
break;
case 'ping':
// Respond with app-level pong
this._sendMessage({ type: 'pong' });
break;
case 'connected':
SyncLog.log(`SuperSyncWebSocketService: Server confirmed connection`);
break;
}
}
private _sendMessage(msg: Record<string, unknown>): void {
if (this._ws?.readyState === WebSocket.OPEN) {
try {
this._ws.send(JSON.stringify(msg));
} catch (err) {
SyncLog.warn('SuperSyncWebSocketService: Failed to send message', err);
}
}
}
private _scheduleReconnect(): void {
if (this._isIntentionalClose || !this._currentParams) {
return;
}
if (this._reconnectAttempts >= MAX_RECONNECT_ATTEMPTS) {
SyncLog.warn(
`SuperSyncWebSocketService: Max reconnect attempts (${MAX_RECONNECT_ATTEMPTS}) reached`,
);
return;
}
this._reconnectAttempts++;
const baseDelay = Math.min(
MIN_RECONNECT_DELAY_MS * Math.pow(2, this._reconnectAttempts - 1),
MAX_RECONNECT_DELAY_MS,
);
// Add jitter to prevent thundering herd
// Random value between -1 and 1 for jitter
const randomFactor = Math.random() * 2 - 1; // eslint-disable-line no-mixed-operators
const jitter = baseDelay * JITTER_FACTOR * randomFactor;
const delay = Math.round(baseDelay + jitter);
SyncLog.log(
`SuperSyncWebSocketService: Reconnecting in ${delay}ms (attempt ${this._reconnectAttempts}/${MAX_RECONNECT_ATTEMPTS})`,
);
this._reconnectTimer = setTimeout(() => {
this._reconnectTimer = null;
if (this._currentParams && !this._isIntentionalClose) {
this._connect(this._currentParams.baseUrl, this._currentParams.accessToken).catch(
(err) => {
SyncLog.warn('SuperSyncWebSocketService: Reconnect attempt failed', err);
this._scheduleReconnect();
},
);
}
}, delay);
}
private _clearReconnectTimer(): void {
if (this._reconnectTimer) {
clearTimeout(this._reconnectTimer);
this._reconnectTimer = null;
}
}
private _startHeartbeatTimer(): void {
this._clearHeartbeatTimer();
this._heartbeatTimer = setTimeout(() => {
SyncLog.warn(
'SuperSyncWebSocketService: No heartbeat received, closing connection',
);
if (this._ws) {
try {
this._ws.close(4000, 'Heartbeat timeout');
} catch (err) {
SyncLog.warn(
'SuperSyncWebSocketService: Error closing on heartbeat timeout',
err,
);
}
}
}, HEARTBEAT_TIMEOUT_MS);
}
private _resetHeartbeatTimer(): void {
if (this._heartbeatTimer) {
this._startHeartbeatTimer();
}
}
private _clearHeartbeatTimer(): void {
if (this._heartbeatTimer) {
clearTimeout(this._heartbeatTimer);
this._heartbeatTimer = null;
}
}
}

View file

@ -0,0 +1,180 @@
import { fakeAsync, flushMicrotasks, TestBed, tick } from '@angular/core/testing';
import { Subject } from 'rxjs';
import {
SuperSyncWebSocketService,
type NewOpsNotification,
} from './super-sync-websocket.service';
import { OperationLogSyncService } from './operation-log-sync.service';
import { SyncProviderManager } from '../sync-providers/provider-manager.service';
import { WrappedProviderService } from '../sync-providers/wrapped-provider.service';
import { WsTriggeredDownloadService } from './ws-triggered-download.service';
import { AuthFailSPError, MissingCredentialsSPError } from '../sync-exports';
describe('WsTriggeredDownloadService', () => {
let service: WsTriggeredDownloadService;
let notification$: Subject<NewOpsNotification>;
let mockWsService: Pick<SuperSyncWebSocketService, 'newOpsNotification$'>;
let mockSyncService: jasmine.SpyObj<OperationLogSyncService>;
let mockProviderManager: jasmine.SpyObj<SyncProviderManager>;
let mockWrappedProvider: jasmine.SpyObj<WrappedProviderService>;
let syncCapableProvider: any;
beforeEach(() => {
notification$ = new Subject<NewOpsNotification>();
mockWsService = {
newOpsNotification$: notification$.asObservable(),
};
syncCapableProvider = { id: 'sync-provider' };
mockSyncService = jasmine.createSpyObj('OperationLogSyncService', [
'downloadRemoteOps',
]);
mockSyncService.downloadRemoteOps.and.returnValue(
Promise.resolve({ kind: 'no_new_ops' as const }),
);
mockProviderManager = jasmine.createSpyObj(
'SyncProviderManager',
['getActiveProvider'],
{
isSyncInProgress: false,
},
);
mockProviderManager.getActiveProvider.and.returnValue({ id: 'raw-provider' } as any);
mockWrappedProvider = jasmine.createSpyObj('WrappedProviderService', [
'getOperationSyncCapable',
]);
mockWrappedProvider.getOperationSyncCapable.and.returnValue(
Promise.resolve(syncCapableProvider as any),
);
TestBed.configureTestingModule({
providers: [
WsTriggeredDownloadService,
{ provide: SuperSyncWebSocketService, useValue: mockWsService },
{ provide: OperationLogSyncService, useValue: mockSyncService },
{ provide: SyncProviderManager, useValue: mockProviderManager },
{ provide: WrappedProviderService, useValue: mockWrappedProvider },
],
});
service = TestBed.inject(WsTriggeredDownloadService);
});
afterEach(() => {
service.stop();
});
it('should trigger a download after the debounce interval', fakeAsync(() => {
service.start();
notification$.next({ latestSeq: 1 });
tick(500);
flushMicrotasks();
expect(mockWrappedProvider.getOperationSyncCapable).toHaveBeenCalled();
expect(mockSyncService.downloadRemoteOps).toHaveBeenCalledWith(syncCapableProvider);
}));
it('should debounce rapid notifications into a single download', fakeAsync(() => {
service.start();
notification$.next({ latestSeq: 1 });
tick(250);
notification$.next({ latestSeq: 2 });
tick(499);
flushMicrotasks();
expect(mockSyncService.downloadRemoteOps).not.toHaveBeenCalled();
tick(1);
flushMicrotasks();
expect(mockSyncService.downloadRemoteOps).toHaveBeenCalledTimes(1);
}));
it('should skip downloads while sync is already in progress', fakeAsync(() => {
mockProviderManager = TestBed.inject(
SyncProviderManager,
) as jasmine.SpyObj<SyncProviderManager>;
Object.defineProperty(mockProviderManager, 'isSyncInProgress', {
get: () => true,
configurable: true,
});
service.start();
notification$.next({ latestSeq: 3 });
tick(500);
flushMicrotasks();
expect(mockWrappedProvider.getOperationSyncCapable).not.toHaveBeenCalled();
expect(mockSyncService.downloadRemoteOps).not.toHaveBeenCalled();
}));
it('should stop listening after an auth failure', fakeAsync(() => {
mockSyncService.downloadRemoteOps.and.callFake(async () => {
throw new AuthFailSPError('unauthorized');
});
service.start();
notification$.next({ latestSeq: 4 });
tick(500);
flushMicrotasks();
notification$.next({ latestSeq: 5 });
tick(500);
flushMicrotasks();
expect(mockSyncService.downloadRemoteOps).toHaveBeenCalledTimes(1);
}));
it('should stop listening after a MissingCredentialsSPError', fakeAsync(() => {
mockSyncService.downloadRemoteOps.and.callFake(async () => {
throw new MissingCredentialsSPError('no creds');
});
service.start();
notification$.next({ latestSeq: 1 });
tick(500);
flushMicrotasks();
notification$.next({ latestSeq: 2 });
tick(500);
flushMicrotasks();
expect(mockSyncService.downloadRemoteOps).toHaveBeenCalledTimes(1);
}));
it('should survive non-auth errors and continue the pipeline', fakeAsync(() => {
let callCount = 0;
mockSyncService.downloadRemoteOps.and.callFake(async () => {
callCount++;
if (callCount === 1) {
throw new Error('network timeout');
}
return { kind: 'no_new_ops' as const };
});
service.start();
notification$.next({ latestSeq: 1 });
tick(500);
flushMicrotasks();
notification$.next({ latestSeq: 2 });
tick(500);
flushMicrotasks();
expect(mockSyncService.downloadRemoteOps).toHaveBeenCalledTimes(2);
}));
it('should be idempotent when start is called twice', fakeAsync(() => {
service.start();
service.start();
notification$.next({ latestSeq: 1 });
tick(500);
flushMicrotasks();
expect(mockSyncService.downloadRemoteOps).toHaveBeenCalledTimes(1);
}));
});

View file

@ -0,0 +1,101 @@
import { inject, Injectable, OnDestroy } from '@angular/core';
import { Subscription } from 'rxjs';
import { debounceTime, exhaustMap, filter } from 'rxjs/operators';
import { SuperSyncWebSocketService } from './super-sync-websocket.service';
import { OperationLogSyncService } from './operation-log-sync.service';
import { SyncProviderManager } from '../sync-providers/provider-manager.service';
import { WrappedProviderService } from '../sync-providers/wrapped-provider.service';
import { SyncLog } from '../../core/log';
import { AuthFailSPError, MissingCredentialsSPError } from '../sync-exports';
const WS_DOWNLOAD_DEBOUNCE_MS = 500;
/**
* Triggers operation downloads when WebSocket notifications arrive.
*
* Pipeline: newOpsNotification$ debounce(500ms) filter(!syncInProgress) exhaustMap(download)
*
* Uses exhaustMap to ignore new notifications while a download is in progress.
* Reuses the existing OperationLogSyncService.downloadRemoteOps() code path.
*/
@Injectable({
providedIn: 'root',
})
export class WsTriggeredDownloadService implements OnDestroy {
private _wsService = inject(SuperSyncWebSocketService);
private _syncService = inject(OperationLogSyncService);
private _providerManager = inject(SyncProviderManager);
private _wrappedProvider = inject(WrappedProviderService);
private _subscription: Subscription | null = null;
start(): void {
if (this._subscription) {
return;
}
this._subscription = this._wsService.newOpsNotification$
.pipe(
debounceTime(WS_DOWNLOAD_DEBOUNCE_MS),
filter(() => !this._providerManager.isSyncInProgress),
exhaustMap((notification) => this._downloadOps(notification.latestSeq)),
)
.subscribe();
SyncLog.log('WsTriggeredDownloadService: Started listening for WS notifications');
}
stop(): void {
this._subscription?.unsubscribe();
this._subscription = null;
SyncLog.log('WsTriggeredDownloadService: Stopped');
}
ngOnDestroy(): void {
this.stop();
}
private async _downloadOps(latestSeq: number): Promise<void> {
if (this._providerManager.isSyncInProgress) {
SyncLog.log('WsTriggeredDownloadService: Sync in progress, skipping WS download');
return;
}
try {
const rawProvider = this._providerManager.getActiveProvider();
if (!rawProvider) {
SyncLog.log(
'WsTriggeredDownloadService: No active provider, skipping WS download',
);
return;
}
const syncCapableProvider =
await this._wrappedProvider.getOperationSyncCapable(rawProvider);
if (!syncCapableProvider) {
SyncLog.log(
'WsTriggeredDownloadService: Provider not operation-sync capable, skipping',
);
return;
}
SyncLog.log(
`WsTriggeredDownloadService: Downloading ops triggered by WS notification (latestSeq=${latestSeq})`,
);
const result = await this._syncService.downloadRemoteOps(syncCapableProvider);
SyncLog.log(`WsTriggeredDownloadService: Download complete. kind=${result.kind}`);
} catch (err) {
if (err instanceof AuthFailSPError || err instanceof MissingCredentialsSPError) {
SyncLog.warn('WsTriggeredDownloadService: Auth failure during download', err);
this.stop();
return;
}
SyncLog.warn(
'WsTriggeredDownloadService: Download failed, periodic sync will retry',
err,
);
}
}
}