Add internal ntfy utility service

Proof: npm test; PYTHONPATH=. python3 test/render_release_manifest_test.py; PYTHONPATH=. python3 test/repo_deployments_test.py; PYTHONPATH=. python3 test/ntfy_manifest_test.py; kubectl kustomize deploy/k8s/base.

Assumptions: ntfy should start as an internal ClusterIP utility so repo-owned services can publish without exposing an unauthenticated public notification endpoint; mobile delivery needs a separate authenticated ingress or external endpoint decision.

Still fake: No public ntfy ingress or operator mobile subscription exists yet; no existing runtime path emits ntfy notifications by default; ntfy cache storage is ephemeral emptyDir.

.forgejo/workflows/deploy.yml

@@ -156,3 +156,5 @@ jobs:
                 kubectl -n "$PROJECT_NAMESPACE" set image "deployment/$deployment" app="$IMAGE"
                 kubectl -n "$PROJECT_NAMESPACE" rollout status "deployment/$deployment" --timeout=180s
               done
+
+          kubectl -n utility rollout status deployment/ntfy --timeout=180s

deploy/k8s/base/kustomization.yaml

@@ -2,6 +2,8 @@ apiVersion: kustomize.config.k8s.io/v1beta1
 kind: Kustomization
 resources:
   - namespace.yaml
+  - utility-namespace.yaml
+  - ntfy.yaml
   - redpanda.yaml
   - postgres.yaml
   - unrip.yaml

deploy/k8s/base/ntfy.yaml

@@ -0,0 +1,86 @@
+apiVersion: v1
+kind: ConfigMap
+metadata:
+  name: ntfy-config
+  namespace: utility
+data:
+  server.yml: |
+    base-url: http://ntfy.utility.svc.cluster.local
+    cache-file: /var/cache/ntfy/cache.db
+    attachment-cache-dir: /var/cache/ntfy/attachments
+---
+apiVersion: v1
+kind: Service
+metadata:
+  name: ntfy
+  namespace: utility
+  labels:
+    app: ntfy
+    app.kubernetes.io/part-of: unrip
+spec:
+  type: ClusterIP
+  selector:
+    app: ntfy
+  ports:
+    - name: http
+      port: 80
+      targetPort: http
+---
+apiVersion: apps/v1
+kind: Deployment
+metadata:
+  name: ntfy
+  namespace: utility
+  labels:
+    app: ntfy
+    app.kubernetes.io/part-of: unrip
+spec:
+  replicas: 1
+  selector:
+    matchLabels:
+      app: ntfy
+  template:
+    metadata:
+      labels:
+        app: ntfy
+        app.kubernetes.io/part-of: unrip
+    spec:
+      containers:
+        - name: ntfy
+          image: binwiederhier/ntfy:v2.21.0
+          imagePullPolicy: IfNotPresent
+          args: ["serve"]
+          ports:
+            - name: http
+              containerPort: 80
+          readinessProbe:
+            httpGet:
+              path: /v1/health
+              port: http
+            initialDelaySeconds: 5
+            periodSeconds: 10
+          livenessProbe:
+            httpGet:
+              path: /v1/health
+              port: http
+            initialDelaySeconds: 20
+            periodSeconds: 30
+          resources:
+            requests:
+              cpu: 25m
+              memory: 64Mi
+            limits:
+              cpu: 250m
+              memory: 128Mi
+          volumeMounts:
+            - name: config
+              mountPath: /etc/ntfy
+              readOnly: true
+            - name: cache
+              mountPath: /var/cache/ntfy
+      volumes:
+        - name: config
+          configMap:
+            name: ntfy-config
+        - name: cache
+          emptyDir: {}

deploy/k8s/base/unrip.yaml

@@ -50,6 +50,9 @@ data:
   STRATEGY_ENGINE_CONTROL_BASE_URL: http://strategy-engine.unrip.svc.cluster.local:8086
   TRADE_EXECUTOR_CONTROL_BASE_URL: http://trade-executor.unrip.svc.cluster.local:8087
   OPS_SENTINEL_CONTROL_BASE_URL: http://ops-sentinel.unrip.svc.cluster.local:8088
+  NOTIFICATION_NTFY_BASE_URL: http://ntfy.utility.svc.cluster.local
+  NOTIFICATION_NTFY_TOPIC: unrip
+  NOTIFICATION_NTFY_TIMEOUT_MS: "5000"
   KAFKA_BROKERS: redpanda.unrip.svc.cluster.local:9092
   KAFKA_CLIENT_ID: unrip
   KAFKA_TOPIC_RAW_NEAR_INTENTS_QUOTE: raw.near_intents.quote

deploy/k8s/base/utility-namespace.yaml

@@ -0,0 +1,7 @@
+apiVersion: v1
+kind: Namespace
+metadata:
+  name: utility
+  labels:
+    app.kubernetes.io/part-of: unrip
+    project.pi.io/type: utility

src/core/notification-client.mjs

@@ -0,0 +1,103 @@
+export function buildNtfyPublishUrl({ baseUrl, topic }) {
+  const normalizedBaseUrl = String(baseUrl || '').trim().replace(/\/+$/, '');
+  const normalizedTopic = String(topic || '').trim();
+  if (!normalizedBaseUrl || !normalizedTopic) return null;
+  return `${normalizedBaseUrl}/${encodeURIComponent(normalizedTopic)}`;
+}
+
+export function buildNtfyPublishRequest({
+  baseUrl,
+  topic,
+  message,
+  title = null,
+  priority = null,
+  tags = [],
+  click = null,
+  token = null,
+} = {}) {
+  const url = buildNtfyPublishUrl({ baseUrl, topic });
+  if (!url) {
+    return { ok: false, skipped: true, reason: 'ntfy_not_configured' };
+  }
+
+  const body = String(message || '').trim();
+  if (!body) {
+    return { ok: false, skipped: true, reason: 'message_empty' };
+  }
+
+  const headers = {
+    'content-type': 'text/plain; charset=utf-8',
+  };
+  if (title) headers.Title = String(title);
+  if (priority) headers.Priority = String(priority);
+  if (tags?.length) headers.Tags = tags.map((tag) => String(tag).trim()).filter(Boolean).join(',');
+  if (click) headers.Click = String(click);
+  if (token) headers.Authorization = `Bearer ${token}`;
+
+  return {
+    ok: true,
+    url,
+    init: {
+      method: 'POST',
+      headers,
+      body,
+    },
+  };
+}
+
+export function createNtfyNotificationClient({
+  baseUrl,
+  topic,
+  token = null,
+  timeoutMs = 5_000,
+  fetchImpl = globalThis.fetch,
+  logger = null,
+} = {}) {
+  return {
+    isConfigured() {
+      return Boolean(buildNtfyPublishUrl({ baseUrl, topic }));
+    },
+
+    async publish(notification = {}) {
+      const request = buildNtfyPublishRequest({
+        baseUrl,
+        topic,
+        token,
+        ...notification,
+      });
+      if (!request.ok) return request;
+      if (typeof fetchImpl !== 'function') {
+        return { ok: false, skipped: true, reason: 'fetch_unavailable' };
+      }
+
+      try {
+        const response = await fetchImpl(request.url, {
+          ...request.init,
+          signal: AbortSignal.timeout(timeoutMs),
+        });
+        const responseText = await response.text().catch(() => '');
+        return {
+          ok: response.ok,
+          status: response.status,
+          response: responseText,
+        };
+      } catch (error) {
+        logger?.error?.('ntfy_notification_failed', {
+          details: {
+            error: {
+              name: error?.name || 'Error',
+              message: error?.message || String(error),
+            },
+          },
+        });
+        return {
+          ok: false,
+          error: {
+            name: error?.name || 'Error',
+            message: error?.message || String(error),
+          },
+        };
+      }
+    },
+  };
+}

src/lib/config.mjs

@@ -105,6 +105,9 @@ const DEFAULTS = {
   operatorDashboardQuoteLimit: 10,
   operatorDashboardTradePageSize: 20,
   operatorDashboardUpstreamTimeoutMs: 3_000,
+  notificationNtfyBaseUrl: '',
+  notificationNtfyTopic: 'unrip',
+  notificationNtfyTimeoutMs: 5_000,
 };
 
 function splitCsv(value) {
@@ -571,5 +574,13 @@ export function loadConfig({ envPath = '.env' } = {}) {
       process.env.OPERATOR_DASHBOARD_UPSTREAM_TIMEOUT_MS,
       DEFAULTS.operatorDashboardUpstreamTimeoutMs,
     ),
+    notificationNtfyBaseUrl:
+      process.env.NOTIFICATION_NTFY_BASE_URL || DEFAULTS.notificationNtfyBaseUrl,
+    notificationNtfyTopic:
+      process.env.NOTIFICATION_NTFY_TOPIC || DEFAULTS.notificationNtfyTopic,
+    notificationNtfyTimeoutMs: parseNumber(
+      process.env.NOTIFICATION_NTFY_TIMEOUT_MS,
+      DEFAULTS.notificationNtfyTimeoutMs,
+    ),
   };
 }

test/notification-client.test.mjs

@@ -0,0 +1,99 @@
+import test from 'node:test';
+import assert from 'node:assert/strict';
+
+import {
+  buildNtfyPublishRequest,
+  buildNtfyPublishUrl,
+  createNtfyNotificationClient,
+} from '../src/core/notification-client.mjs';
+import { loadConfig } from '../src/lib/config.mjs';
+
+test('ntfy publish request uses repo configured base url and topic', () => {
+  const request = buildNtfyPublishRequest({
+    baseUrl: 'http://ntfy.utility.svc.cluster.local/',
+    topic: 'unrip alerts',
+    title: 'Quote lifecycle',
+    message: 'quote abc submitted, awaiting settlement',
+    priority: 'high',
+    tags: ['warning', 'unrip'],
+    click: 'https://dashboard.example/strategy',
+  });
+
+  assert.equal(request.ok, true);
+  assert.equal(request.url, 'http://ntfy.utility.svc.cluster.local/unrip%20alerts');
+  assert.equal(request.init.method, 'POST');
+  assert.equal(request.init.body, 'quote abc submitted, awaiting settlement');
+  assert.equal(request.init.headers.Title, 'Quote lifecycle');
+  assert.equal(request.init.headers.Priority, 'high');
+  assert.equal(request.init.headers.Tags, 'warning,unrip');
+  assert.equal(request.init.headers.Click, 'https://dashboard.example/strategy');
+});
+
+test('ntfy notification client skips when endpoint or message is not configured', async () => {
+  assert.equal(buildNtfyPublishUrl({ baseUrl: '', topic: 'unrip' }), null);
+  assert.deepEqual(
+    buildNtfyPublishRequest({ baseUrl: 'http://ntfy.utility.svc.cluster.local', topic: 'unrip', message: '' }),
+    { ok: false, skipped: true, reason: 'message_empty' },
+  );
+
+  const client = createNtfyNotificationClient({ baseUrl: '', topic: 'unrip' });
+  assert.equal(client.isConfigured(), false);
+  assert.deepEqual(await client.publish({ message: 'hello' }), {
+    ok: false,
+    skipped: true,
+    reason: 'ntfy_not_configured',
+  });
+});
+
+test('ntfy notification client posts text and returns response status', async () => {
+  const requests = [];
+  const client = createNtfyNotificationClient({
+    baseUrl: 'http://ntfy.utility.svc.cluster.local',
+    topic: 'unrip',
+    timeoutMs: 1234,
+    fetchImpl: async (url, init) => {
+      requests.push({ url, init });
+      return {
+        ok: true,
+        status: 200,
+        async text() { return '{"id":"msg-1"}'; },
+      };
+    },
+  });
+
+  const result = await client.publish({ message: 'request completed', title: 'unrip' });
+
+  assert.equal(result.ok, true);
+  assert.equal(result.status, 200);
+  assert.equal(requests[0].url, 'http://ntfy.utility.svc.cluster.local/unrip');
+  assert.equal(requests[0].init.body, 'request completed');
+  assert.equal(requests[0].init.headers.Title, 'unrip');
+  assert.ok(requests[0].init.signal);
+});
+
+test('config exposes ntfy notification defaults and environment overrides', () => {
+  const previous = {
+    baseUrl: process.env.NOTIFICATION_NTFY_BASE_URL,
+    topic: process.env.NOTIFICATION_NTFY_TOPIC,
+    timeout: process.env.NOTIFICATION_NTFY_TIMEOUT_MS,
+  };
+  process.env.NOTIFICATION_NTFY_BASE_URL = 'http://ntfy.utility.svc.cluster.local';
+  process.env.NOTIFICATION_NTFY_TOPIC = 'unrip';
+  process.env.NOTIFICATION_NTFY_TIMEOUT_MS = '4321';
+
+  try {
+    const config = loadConfig({ envPath: '/tmp/unrip-missing-env-for-notification-test' });
+    assert.equal(config.notificationNtfyBaseUrl, 'http://ntfy.utility.svc.cluster.local');
+    assert.equal(config.notificationNtfyTopic, 'unrip');
+    assert.equal(config.notificationNtfyTimeoutMs, 4321);
+  } finally {
+    restoreEnv('NOTIFICATION_NTFY_BASE_URL', previous.baseUrl);
+    restoreEnv('NOTIFICATION_NTFY_TOPIC', previous.topic);
+    restoreEnv('NOTIFICATION_NTFY_TIMEOUT_MS', previous.timeout);
+  }
+});
+
+function restoreEnv(key, value) {
+  if (value == null) delete process.env[key];
+  else process.env[key] = value;
+}

test/ntfy_manifest_test.py

@@ -0,0 +1,35 @@
+import pathlib
+import re
+import unittest
+
+ROOT = pathlib.Path(__file__).resolve().parents[1]
+
+
+class NtfyManifestTest(unittest.TestCase):
+    def test_kustomization_includes_internal_ntfy_utility_resources(self):
+        source = (ROOT / 'deploy/k8s/base/kustomization.yaml').read_text()
+        self.assertIn('utility-namespace.yaml', source)
+        self.assertIn('ntfy.yaml', source)
+
+    def test_ntfy_manifest_is_internal_clusterip_service_with_health_checks(self):
+        source = (ROOT / 'deploy/k8s/base/ntfy.yaml').read_text()
+        self.assertIn('namespace: utility', source)
+        self.assertIn('image: binwiederhier/ntfy:v2.21.0', source)
+        self.assertRegex(source, r'kind: Service[\s\S]*type: ClusterIP')
+        self.assertIn('path: /v1/health', source)
+        self.assertIn('base-url: http://ntfy.utility.svc.cluster.local', source)
+        self.assertNotIn('kind: Ingress', source)
+
+    def test_unrip_services_receive_internal_notification_endpoint(self):
+        source = (ROOT / 'deploy/k8s/base/unrip.yaml').read_text()
+        self.assertIn('NOTIFICATION_NTFY_BASE_URL: http://ntfy.utility.svc.cluster.local', source)
+        self.assertIn('NOTIFICATION_NTFY_TOPIC: unrip', source)
+
+    def test_workflow_waits_for_ntfy_rollout_without_rewriting_external_image(self):
+        source = (ROOT / '.forgejo/workflows/deploy.yml').read_text()
+        self.assertIn('kubectl -n utility rollout status deployment/ntfy --timeout=180s', source)
+        self.assertNotRegex(source, re.compile(r'set image "deployment/ntfy"'))
+
+
+if __name__ == '__main__':
+    unittest.main()