feat: zentrale selfhost-dienste bereitstellen

push-server/apps/api/Dockerfile

@@ -12,11 +12,13 @@ RUN npm run build --workspace @fedeo/push-db && npm run build --workspace @fedeo
 FROM node:22-alpine
 WORKDIR /app
 ENV NODE_ENV=production
+ENV DRIZZLE_MIGRATIONS_FOLDER=/app/packages/db/drizzle
 COPY --from=deps /app/node_modules ./node_modules
 COPY --from=build /app/apps/api/dist ./apps/api/dist
 COPY --from=build /app/packages/db/dist ./packages/db/dist
+COPY --from=build /app/packages/db/drizzle ./packages/db/drizzle
 COPY --from=build /app/package.json ./package.json
 COPY --from=build /app/apps/api/package.json ./apps/api/package.json
 COPY --from=build /app/packages/db/package.json ./packages/db/package.json
 EXPOSE 4020
-CMD ["node", "apps/api/dist/index.js"]
+CMD ["sh", "-c", "node packages/db/dist/src/migrate.js && node apps/api/dist/index.js"]

push-server/apps/api/src/config/env.ts

@@ -25,6 +25,11 @@ const envSchema = z.object({
   ANDROID_SENDER_ID: z.string().optional().default(""),
   FCM_PROJECT_ID: z.string().optional().default(""),
   FCM_SERVICE_ACCOUNT_JSON: z.string().optional().default(""),
+  OPENAI_API_KEY: z.string().optional().default(""),
+  OPENAI_BASE_URL: z.string().url().default("https://api.openai.com/v1"),
+  GOCARDLESS_BASE_URL: z.string().url().default("https://bankaccountdata.gocardless.com/api/v2"),
+  GOCARDLESS_SECRET_ID: z.string().optional().default(""),
+  GOCARDLESS_SECRET_KEY: z.string().optional().default(""),
 });
 
 export const env = envSchema.parse(process.env);

push-server/apps/api/src/index.ts

@@ -7,6 +7,7 @@ import { pool } from "./db/client.js";
 import { adminRoutes } from "./routes/admin.js";
 import { instanceRoutes } from "./routes/instance.js";
 import { publicRoutes } from "./routes/public.js";
+import { serviceRoutes } from "./routes/services.js";
 
 const app = Fastify({
   logger: true,
@@ -39,6 +40,7 @@ await app.register(cors, {
 await app.register(publicRoutes);
 await app.register(adminRoutes);
 await app.register(instanceRoutes);
+await app.register(serviceRoutes);
 
 const close = async () => {
   await app.close();

push-server/apps/api/src/routes/admin.ts

@@ -1,7 +1,7 @@
 import type { FastifyInstance } from "fastify";
 import { and, desc, eq, sql } from "drizzle-orm";
 import { z } from "zod";
-import { auditLogs, deliveryJobs, pushDevices, pushInstances } from "@fedeo/push-db";
+import { auditLogs, deliveryJobs, pushDevices, pushInstances, serviceEntitlements, serviceUsageEvents } from "@fedeo/push-db";
 import { db } from "../db/client.js";
 import { requireAdmin } from "../lib/auth.js";
 import { encryptSecret } from "../lib/crypto.js";
@@ -29,6 +29,13 @@ const testPushSchema = z.object({
   priority: z.enum(["normal", "high"]).default("high"),
 });
 
+const entitlementSchema = z.object({
+  enabled: z.boolean(),
+  monthlyLimit: z.number().int().positive().nullable().default(null),
+  unitPriceMicros: z.number().int().nonnegative().default(0),
+  metadata: z.record(z.string(), z.unknown()).default({}),
+});
+
 export async function adminRoutes(app: FastifyInstance): Promise<void> {
   app.addHook("preHandler", requireAdmin);
 
@@ -37,11 +44,13 @@ export async function adminRoutes(app: FastifyInstance): Promise<void> {
     const [devices] = await db.select({ count: sql<number>`count(*)::int` }).from(pushDevices);
     const [jobs] = await db.select({ count: sql<number>`count(*)::int` }).from(deliveryJobs);
     const [failedJobs] = await db.select({ count: sql<number>`count(*)::int` }).from(deliveryJobs).where(eq(deliveryJobs.status, "failed"));
+    const [usage] = await db.select({ units: sql<number>`coalesce(sum(${serviceUsageEvents.units}), 0)::bigint` }).from(serviceUsageEvents).where(eq(serviceUsageEvents.status, "succeeded"));
     return {
       instances: instances?.count || 0,
       devices: devices?.count || 0,
       jobs: jobs?.count || 0,
       failedJobs: failedJobs?.count || 0,
+      serviceUnits: Number(usage?.units || 0),
     };
   });
 
@@ -130,6 +139,44 @@ export async function adminRoutes(app: FastifyInstance): Promise<void> {
     return await db.select().from(deliveryJobs).where(eq(deliveryJobs.instanceId, params.id)).orderBy(desc(deliveryJobs.createdAt)).limit(100);
   });
 
+  app.get("/admin/instances/:id/services", async (request) => {
+    const params = z.object({ id: z.string().uuid() }).parse(request.params);
+    return db.select().from(serviceEntitlements).where(eq(serviceEntitlements.instanceId, params.id));
+  });
+
+  app.put("/admin/instances/:id/services/:service", async (request, reply) => {
+    const params = z.object({ id: z.string().uuid(), service: z.enum(["ai", "banking"]) }).parse(request.params);
+    const body = entitlementSchema.parse(request.body);
+    const [instance] = await db.select({ id: pushInstances.id }).from(pushInstances).where(eq(pushInstances.id, params.id)).limit(1);
+    if (!instance) return reply.code(404).send({ error: "instance_not_found" });
+    const [saved] = await db.insert(serviceEntitlements).values({
+      instanceId: params.id,
+      service: params.service,
+      ...body,
+    }).onConflictDoUpdate({
+      target: [serviceEntitlements.instanceId, serviceEntitlements.service],
+      set: { ...body, updatedAt: new Date() },
+    }).returning();
+    await audit("admin", "instance.service.updated", params.id, { service: params.service, enabled: body.enabled, monthlyLimit: body.monthlyLimit });
+    return saved;
+  });
+
+  app.get("/admin/instances/:id/usage", async (request) => {
+    const params = z.object({ id: z.string().uuid() }).parse(request.params);
+    const query = z.object({ limit: z.coerce.number().int().positive().max(1000).default(200) }).parse(request.query);
+    return db.select().from(serviceUsageEvents).where(eq(serviceUsageEvents.instanceId, params.id)).orderBy(desc(serviceUsageEvents.createdAt)).limit(query.limit);
+  });
+
+  app.get("/admin/usage/summary", async () => {
+    return db.select({
+      instanceId: serviceUsageEvents.instanceId,
+      service: serviceUsageEvents.service,
+      units: sql<number>`coalesce(sum(${serviceUsageEvents.units}), 0)::bigint`,
+      costMicros: sql<number>`coalesce(sum(${serviceUsageEvents.costMicros}), 0)::bigint`,
+      calls: sql<number>`count(*)::int`,
+    }).from(serviceUsageEvents).where(eq(serviceUsageEvents.status, "succeeded")).groupBy(serviceUsageEvents.instanceId, serviceUsageEvents.service);
+  });
+
   app.post("/admin/instances/:id/test-push", async (request, reply) => {
     const params = z.object({ id: z.string().uuid() }).parse(request.params);
     const body = testPushSchema.parse(request.body || {});

push-server/apps/api/src/routes/public.ts

@@ -4,13 +4,13 @@ import { env } from "../config/env.js";
 export async function publicRoutes(app: FastifyInstance): Promise<void> {
   app.get("/health", async () => ({
     status: "ok",
-    service: "fedeo-push-api",
+    service: "fedeo-central-services-api",
   }));
 
   app.get("/v1/public-config", async () => ({
     webPushPublicKey: env.WEB_PUSH_PUBLIC_KEY || null,
     iosBundleId: env.IOS_BUNDLE_ID,
     androidSenderId: env.ANDROID_SENDER_ID || null,
-    capabilities: ["ios_push", "minimal_payload", "instance_hmac"],
+    capabilities: ["ios_push", "minimal_payload", "instance_hmac", "service_entitlements", "usage_metering", "central_ai", "central_banking"],
   }));
 }

push-server/apps/api/src/routes/services.ts

@@ -0,0 +1,131 @@
+import type { FastifyInstance, FastifyRequest } from "fastify";
+import { z } from "zod";
+import { env } from "../config/env.js";
+import { requireInstance } from "../lib/auth.js";
+import { recordServiceUsage, requireCentralService } from "../services/central-services.js";
+
+const aiSchema = z.object({
+  model: z.enum(["gpt-4o", "gpt-4o-mini"]),
+  messages: z.array(z.object({
+    role: z.enum(["system", "user", "assistant"]),
+    content: z.union([z.string(), z.array(z.unknown())]),
+  }).passthrough()).min(1).max(100),
+  response_format: z.record(z.string(), z.unknown()).optional(),
+  temperature: z.number().min(0).max(2).optional(),
+  max_tokens: z.number().int().positive().max(16_384).optional(),
+}).passthrough();
+
+const requisitionSchema = z.object({
+  institutionId: z.string().min(1).max(200),
+  redirect: z.string().url(),
+  userLanguage: z.string().length(2).default("de"),
+});
+
+let bankingToken: { access: string; expiresAt: number } | null = null;
+
+async function providerJson(url: string, init: RequestInit) {
+  const response = await fetch(url, init);
+  const text = await response.text();
+  let data: any = null;
+  try { data = text ? JSON.parse(text) : null; } catch { data = { message: text }; }
+  if (!response.ok) {
+    const error = new Error(data?.detail || data?.message || `Provider antwortete mit HTTP ${response.status}`) as Error & { code?: string; status?: number };
+    error.code = data?.code || `provider_http_${response.status}`;
+    error.status = response.status;
+    throw error;
+  }
+  return { data, response };
+}
+
+async function getBankingToken() {
+  if (bankingToken && bankingToken.expiresAt > Date.now() + 60_000) return bankingToken.access;
+  if (!env.GOCARDLESS_SECRET_ID || !env.GOCARDLESS_SECRET_KEY) throw Object.assign(new Error("GoCardless ist zentral nicht konfiguriert."), { code: "banking_provider_not_configured" });
+  const { data } = await providerJson(`${env.GOCARDLESS_BASE_URL.replace(/\/$/, "")}/token/new/`, {
+    method: "POST",
+    headers: { "Content-Type": "application/json" },
+    body: JSON.stringify({ secret_id: env.GOCARDLESS_SECRET_ID, secret_key: env.GOCARDLESS_SECRET_KEY }),
+  });
+  bankingToken = { access: data.access, expiresAt: Date.now() + Number(data.access_expires || 86_400) * 1000 };
+  return bankingToken.access;
+}
+
+async function bankingRequest(path: string, init: RequestInit = {}) {
+  const access = await getBankingToken();
+  return providerJson(`${env.GOCARDLESS_BASE_URL.replace(/\/$/, "")}${path}`, {
+    ...init,
+    headers: { Accept: "application/json", Authorization: `Bearer ${access}`, ...(init.headers || {}) },
+  });
+}
+
+function safeProviderError(error: any) {
+  return { error: error?.code || "provider_error", message: error?.message || "Zentraler Provider-Aufruf fehlgeschlagen." };
+}
+
+async function withBankingUsage(request: FastifyRequest, reply: any, operation: string, handler: () => Promise<{ data: any; response: Response }>) {
+  const entitlement = await requireCentralService(request, reply, "banking");
+  if (!entitlement) return;
+  try {
+    const result = await handler();
+    const requestId = await recordServiceUsage({ request, service: "banking", operation, status: "succeeded", unitPriceMicros: entitlement.unitPriceMicros, provider: "gocardless", providerRequestId: result.response.headers.get("x-request-id") });
+    reply.header("X-Fedeo-Request-Id", requestId);
+    return result.data;
+  } catch (error: any) {
+    await recordServiceUsage({ request, service: "banking", operation, status: "failed", units: 0, unitPriceMicros: entitlement.unitPriceMicros, provider: "gocardless", errorCode: error?.code });
+    return reply.code(error?.status && error.status < 500 ? error.status : 502).send(safeProviderError(error));
+  }
+}
+
+export async function serviceRoutes(app: FastifyInstance): Promise<void> {
+  app.addHook("preHandler", requireInstance);
+
+  app.post("/v1/services/ai/chat-completions", async (request, reply) => {
+    const entitlement = await requireCentralService(request, reply, "ai");
+    if (!entitlement) return;
+    if (!env.OPENAI_API_KEY) return reply.code(503).send({ error: "ai_provider_not_configured", message: "OpenAI ist zentral nicht konfiguriert." });
+    const body = aiSchema.parse(request.body);
+    try {
+      const { data, response } = await providerJson(`${env.OPENAI_BASE_URL.replace(/\/$/, "")}/chat/completions`, {
+        method: "POST",
+        headers: { "Content-Type": "application/json", Authorization: `Bearer ${env.OPENAI_API_KEY}` },
+        body: JSON.stringify(body),
+      });
+      const units = Number(data?.usage?.total_tokens || 0);
+      const requestId = await recordServiceUsage({ request, service: "ai", operation: "chat_completions", units, status: "succeeded", unitPriceMicros: entitlement.unitPriceMicros, provider: "openai", providerRequestId: response.headers.get("x-request-id"), metadata: { model: body.model, promptTokens: data?.usage?.prompt_tokens || 0, completionTokens: data?.usage?.completion_tokens || 0 } });
+      reply.header("X-Fedeo-Request-Id", requestId);
+      return data;
+    } catch (error: any) {
+      await recordServiceUsage({ request, service: "ai", operation: "chat_completions", units: 0, status: "failed", unitPriceMicros: entitlement.unitPriceMicros, provider: "openai", errorCode: error?.code, metadata: { model: body.model } });
+      return reply.code(error?.status && error.status < 500 ? error.status : 502).send(safeProviderError(error));
+    }
+  });
+
+  app.get("/v1/services/banking/institutions", async (request, reply) => {
+    const query = z.object({ country: z.string().length(2).default("DE") }).parse(request.query);
+    return withBankingUsage(request, reply, "institutions", () => bankingRequest(`/institutions/?country=${encodeURIComponent(query.country.toLowerCase())}`));
+  });
+
+  app.post("/v1/services/banking/requisitions", async (request, reply) => {
+    const body = requisitionSchema.parse(request.body);
+    if (new URL(body.redirect).origin !== new URL(request.pushInstance!.baseUrl).origin) {
+      return reply.code(400).send({ error: "redirect_origin_invalid", message: "Banking-Redirect muss zur registrierten Instanz gehören." });
+    }
+    return withBankingUsage(request, reply, "create_requisition", () => bankingRequest("/requisitions/", { method: "POST", headers: { "Content-Type": "application/json" }, body: JSON.stringify({ redirect: body.redirect, institution_id: body.institutionId, user_language: body.userLanguage }) }));
+  });
+
+  app.get("/v1/services/banking/requisitions/:id", async (request, reply) => {
+    const { id } = z.object({ id: z.string().min(1).max(200) }).parse(request.params);
+    return withBankingUsage(request, reply, "get_requisition", () => bankingRequest(`/requisitions/${encodeURIComponent(id)}/`));
+  });
+
+  app.get("/v1/services/banking/accounts/:id", async (request, reply) => {
+    const { id } = z.object({ id: z.string().min(1).max(200) }).parse(request.params);
+    return withBankingUsage(request, reply, "get_account", () => bankingRequest(`/accounts/${encodeURIComponent(id)}/`));
+  });
+
+  for (const resource of ["balances", "transactions"] as const) {
+    app.get(`/v1/services/banking/accounts/:id/${resource}`, async (request, reply) => {
+      const { id } = z.object({ id: z.string().min(1).max(200) }).parse(request.params);
+      return withBankingUsage(request, reply, `get_${resource}`, () => bankingRequest(`/accounts/${encodeURIComponent(id)}/${resource}/`));
+    });
+  }
+}

push-server/apps/api/src/services/central-services.ts

@@ -0,0 +1,84 @@
+import { randomUUID } from "node:crypto";
+import type { FastifyReply, FastifyRequest } from "fastify";
+import { and, eq, gte, sql } from "drizzle-orm";
+import { serviceEntitlements, serviceUsageEvents } from "@fedeo/push-db";
+import { db } from "../db/client.js";
+
+export type CentralServiceName = "ai" | "banking";
+
+export async function requireCentralService(
+  request: FastifyRequest,
+  reply: FastifyReply,
+  service: CentralServiceName,
+) {
+  const instance = request.pushInstance!;
+  const [entitlement] = await db
+    .select()
+    .from(serviceEntitlements)
+    .where(and(eq(serviceEntitlements.instanceId, instance.id), eq(serviceEntitlements.service, service)))
+    .limit(1);
+
+  if (!entitlement?.enabled) {
+    await reply.code(403).send({
+      error: "service_not_enabled",
+      message: `Der zentrale Dienst '${service}' ist für diese Instanz nicht freigeschaltet.`,
+    });
+    return null;
+  }
+
+  if (entitlement.monthlyLimit !== null) {
+    const monthStart = new Date();
+    monthStart.setUTCDate(1);
+    monthStart.setUTCHours(0, 0, 0, 0);
+    const [usage] = await db
+      .select({ units: sql<number>`coalesce(sum(${serviceUsageEvents.units}), 0)::bigint` })
+      .from(serviceUsageEvents)
+      .where(and(
+        eq(serviceUsageEvents.instanceId, instance.id),
+        eq(serviceUsageEvents.service, service),
+        eq(serviceUsageEvents.status, "succeeded"),
+        gte(serviceUsageEvents.createdAt, monthStart),
+      ));
+    if (Number(usage?.units || 0) >= entitlement.monthlyLimit) {
+      await reply.code(429).send({
+        error: "service_monthly_limit_reached",
+        message: `Das Monatslimit für den zentralen Dienst '${service}' ist erreicht.`,
+        limit: entitlement.monthlyLimit,
+        used: Number(usage?.units || 0),
+      });
+      return null;
+    }
+  }
+
+  return entitlement;
+}
+
+export async function recordServiceUsage(input: {
+  request: FastifyRequest;
+  service: CentralServiceName;
+  operation: string;
+  units?: number;
+  status: "succeeded" | "failed";
+  unitPriceMicros: number;
+  provider?: string;
+  providerRequestId?: string | null;
+  errorCode?: string;
+  metadata?: Record<string, unknown>;
+}) {
+  const requestId = String(input.request.headers["x-fedeo-request-id"] || randomUUID());
+  const units = Math.max(0, Math.trunc(input.units ?? 1));
+  await db.insert(serviceUsageEvents).values({
+    requestId,
+    instanceId: input.request.pushInstance!.id,
+    service: input.service,
+    operation: input.operation,
+    units,
+    costMicros: input.status === "succeeded" ? units * input.unitPriceMicros : 0,
+    status: input.status,
+    provider: input.provider,
+    providerRequestId: input.providerRequestId || null,
+    errorCode: input.errorCode,
+    metadata: input.metadata || {},
+  });
+  return requestId;
+}