Upload integral_apollo_server.js

integral_apollo_server.js

@@ -0,0 +1,541 @@
+// Integral API — Apollo Server implementation (MVP)
+// This document contains multiple files you can copy into a project.
+
+/*
+File structure (paste into repo):
+
+integral-apollo-server/
+├─ package.json
+├─ .env.example
+├─ docker-compose.yml
+├─ migrations/01_schema.sql
+├─ src/
+|  ├─ index.js          -> Apollo server entry
+|  ├─ db.js             -> Postgres pool
+|  ├─ pubsub.js         -> Redis-backed PubSub
+|  ├─ schema.js         -> GraphQL typeDefs
+|  └─ resolvers.js      -> resolvers wired to Postgres + PubSub
+└─ README.md
+
+Follow README to run.
+*/
+
+// -------------------------
+// package.json
+// -------------------------
+
+/*
+{
+  "name": "integral-apollo-server",
+  "version": "0.1.0",
+  "main": "src/index.js",
+  "scripts": {
+    "start": "node src/index.js",
+    "dev": "nodemon --watch src --exec node src/index.js"
+  },
+  "engines": { "node": ">=18" },
+  "dependencies": {
+    "apollo-server": "^3.11.1",
+    "graphql": "^16.6.0",
+    "pg": "^8.11.0",
+    "ioredis": "^5.3.2",
+    "graphql-redis-subscriptions": "^3.1.0",
+    "uuid": "^9.0.0",
+    "dotenv": "^16.0.3"
+  },
+  "devDependencies": {
+    "nodemon": "^2.0.22"
+  }
+}
+*/
+
+// -------------------------
+// .env.example
+// -------------------------
+
+/*
+# Postgres
+DATABASE_URL=postgresql://integral:integral@postgres:5432/integral
+
+# Redis
+REDIS_URL=redis://redis:6379
+
+# Server
+PORT=4000
+
+# JWT (if you add auth later)
+JWT_SECRET=changeme
+*/
+
+// -------------------------
+// docker-compose.yml
+// -------------------------
+
+/*
+version: '3.8'
+services:
+  postgres:
+    image: postgres:15
+    environment:
+      POSTGRES_DB: integral
+      POSTGRES_USER: integral
+      POSTGRES_PASSWORD: integral
+    volumes:
+      - pgdata:/var/lib/postgresql/data
+    ports:
+      - '5432:5432'
+
+  redis:
+    image: redis:7
+    command: ["redis-server", "--save", "60", "1", "--appendonly", "yes"]
+    ports:
+      - '6379:6379'
+
+  app:
+    build: .
+    image: integral-apollo-server:dev
+    depends_on:
+      - postgres
+      - redis
+    environment:
+      - DATABASE_URL=${DATABASE_URL}
+      - REDIS_URL=${REDIS_URL}
+      - PORT=${PORT}
+    volumes:
+      - ./:/usr/src/app
+    command: ["npm", "run", "dev"]
+    ports:
+      - '4000:4000'
+
+volumes:
+  pgdata:
+*/
+
+// -------------------------
+// migrations/01_schema.sql
+// -------------------------
+
+/*
+-- migrations/01_schema.sql
+CREATE EXTENSION IF NOT EXISTS pgcrypto;
+
+CREATE TABLE IF NOT EXISTS objects (
+  id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
+  namespace TEXT NOT NULL,
+  type TEXT NOT NULL,
+  timestamp TIMESTAMPTZ NOT NULL,
+  location JSONB,
+  severity INT,
+  confirmed BOOLEAN DEFAULT FALSE,
+  images JSONB,
+  provenance JSONB,
+  created_at TIMESTAMPTZ DEFAULT now()
+);
+
+CREATE TABLE IF NOT EXISTS payouts (
+  id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
+  fault_id UUID REFERENCES objects(id) ON DELETE CASCADE,
+  amount_minor_units BIGINT NOT NULL,
+  currency VARCHAR(8) NOT NULL DEFAULT 'ZAR',
+  payee_id TEXT NOT NULL,
+  status VARCHAR(32) NOT NULL DEFAULT 'created',
+  tx_ref TEXT,
+  created_at TIMESTAMPTZ DEFAULT now(),
+  settled_at TIMESTAMPTZ
+);
+
+CREATE INDEX IF NOT EXISTS idx_objects_namespace ON objects(namespace);
+CREATE INDEX IF NOT EXISTS idx_objects_confirmed ON objects(confirmed);
+*/
+
+// -------------------------
+// src/db.js
+// -------------------------
+
+/* src/db.js */
+const { Pool } = require('pg');
+require('dotenv').config();
+
+const pool = new Pool({ connectionString: process.env.DATABASE_URL });
+
+module.exports = {
+  query: (text, params) => pool.query(text, params),
+  pool
+};
+
+// -------------------------
+// src/pubsub.js
+// -------------------------
+
+/* src/pubsub.js */
+const { RedisPubSub } = require('graphql-redis-subscriptions');
+const Redis = require('ioredis');
+require('dotenv').config();
+
+const options = {
+  retryStrategy: times => Math.min(times * 50, 2000),
+};
+
+const pubsub = new RedisPubSub({
+  publisher: new Redis(process.env.REDIS_URL, options),
+  subscriber: new Redis(process.env.REDIS_URL, options),
+});
+
+module.exports = pubsub;
+
+// -------------------------
+// src/schema.js
+// -------------------------
+
+/* src/schema.js */
+const { gql } = require('apollo-server');
+
+const typeDefs = gql`
+scalar DateTime
+scalar JSON
+
+enum Namespace { infrastructure satellite }
+
+type Location { lat: Float lon: Float }
+
+type Provenance { source: String! license: String retrieved_at: DateTime! }
+
+type InfrastructureFault {
+  id: ID!
+  namespace: Namespace!
+  type: String!
+  timestamp: DateTime!
+  location: JSON
+  severity: Int!
+  confirmed: Boolean!
+  images: [String]
+  provenance: Provenance
+}
+
+type Payout {
+  id: ID!
+  faultId: ID!
+  amountMinorUnits: Int!
+  currency: String!
+  payeeId: String!
+  status: String!
+  createdAt: DateTime!
+  settledAt: DateTime
+  txRef: String
+}
+
+input IngestFaultInput {
+  namespace: Namespace!
+  type: String!
+  timestamp: DateTime!
+  location: JSON
+  severity: Int!
+  images: [String]
+  provenance: JSON
+}
+
+input CreatePayoutInput {
+  faultId: ID!
+  amountMinorUnits: Int!
+  currency: String!
+  payeeId: String!
+}
+
+type Query {
+  listInfraFaults(limit: Int = 50, offset: Int = 0): [InfrastructureFault!]
+  payoutsForFault(faultId: ID!): [Payout!]
+}
+
+type Mutation {
+  ingestFault(input: IngestFaultInput!): InfrastructureFault!
+  confirmFault(id: ID!, confirmed: Boolean!): InfrastructureFault!
+  createPayout(input: CreatePayoutInput!): Payout!
+  settlePayout(payoutId: ID!): Payout!
+}
+
+type Subscription {
+  faultCreated: InfrastructureFault!
+  faultConfirmed: InfrastructureFault!
+  payoutUpdated: Payout!
+}
+`;
+
+module.exports = typeDefs;
+
+// -------------------------
+// src/resolvers.js
+// -------------------------
+
+/* src/resolvers.js */
+const { GraphQLScalarType, Kind } = require('graphql');
+const db = require('./db');
+const pubsub = require('./pubsub');
+const { v4: uuidv4 } = require('uuid');
+
+const FAULT_CREATED = 'FAULT_CREATED';
+const FAULT_CONFIRMED = 'FAULT_CONFIRMED';
+const PAYOUT_UPDATED = 'PAYOUT_UPDATED';
+
+const DateTime = new GraphQLScalarType({
+  name: 'DateTime',
+  description: 'ISO date-time scalar',
+  parseValue: value => new Date(value),
+  serialize: value => value instanceof Date ? value.toISOString() : new Date(value).toISOString(),
+  parseLiteral(ast) {
+    if (ast.kind === Kind.STRING) return new Date(ast.value);
+    return null;
+  }
+});
+
+const JSONScalar = new GraphQLScalarType({
+  name: 'JSON',
+  description: 'Arbitrary JSON value',
+  parseValue: value => value,
+  serialize: value => value,
+  parseLiteral(ast) {
+    switch (ast.kind) {
+      case Kind.STRING: return ast.value;
+      case Kind.INT: return parseInt(ast.value, 10);
+      case Kind.FLOAT: return parseFloat(ast.value);
+      case Kind.BOOLEAN: return ast.value === 'true';
+      default: return null;
+    }
+  }
+});
+
+const resolvers = {
+  DateTime,
+  JSON: JSONScalar,
+
+  Query: {
+    listInfraFaults: async (_, { limit, offset }) => {
+      const r = await db.query('SELECT * FROM objects WHERE type=$1 ORDER BY created_at DESC LIMIT $2 OFFSET $3', ['pothole-detection', limit, offset]);
+      return r.rows.map(r => ({
+        id: r.id,
+        namespace: r.namespace,
+        type: r.type,
+        timestamp: r.timestamp,
+        location: r.location,
+        severity: r.severity,
+        confirmed: r.confirmed,
+        images: r.images,
+        provenance: r.provenance
+      }));
+    },
+    payoutsForFault: async (_, { faultId }) => {
+      const r = await db.query('SELECT * FROM payouts WHERE fault_id=$1 ORDER BY created_at DESC', [faultId]);
+      return r.rows.map(p => ({
+        id: p.id,
+        faultId: p.fault_id,
+        amountMinorUnits: parseInt(p.amount_minor_units, 10),
+        currency: p.currency,
+        payeeId: p.payee_id,
+        status: p.status,
+        createdAt: p.created_at,
+        settledAt: p.settled_at,
+        txRef: p.tx_ref
+      }));
+    }
+  },
+
+  Mutation: {
+    ingestFault: async (_, { input }) => {
+      const id = uuidv4();
+      const q = `INSERT INTO objects (id, namespace, type, timestamp, location, severity, images, provenance)
+                 VALUES ($1,$2,$3,$4,$5,$6,$7,$8) RETURNING *`;
+      const vals = [id, input.namespace, input.type, input.timestamp || new Date().toISOString(), JSON.stringify(input.location || null), input.severity, JSON.stringify(input.images || []), input.provenance || {}];
+      const r = await db.query(q, vals);
+      const obj = r.rows[0];
+      const payload = {
+        id: obj.id,
+        namespace: obj.namespace,
+        type: obj.type,
+        timestamp: obj.timestamp,
+        location: obj.location,
+        severity: obj.severity,
+        confirmed: obj.confirmed,
+        images: obj.images,
+        provenance: obj.provenance
+      };
+      await pubsub.publish(FAULT_CREATED, payload);
+      return payload;
+    },
+
+    confirmFault: async (_, { id, confirmed }) => {
+      const q = 'UPDATE objects SET confirmed=$1 WHERE id=$2 RETURNING *';
+      const r = await db.query(q, [confirmed, id]);
+      if (r.rowCount === 0) throw new Error('fault not found');
+      const obj = r.rows[0];
+      const payload = {
+        id: obj.id,
+        namespace: obj.namespace,
+        type: obj.type,
+        timestamp: obj.timestamp,
+        location: obj.location,
+        severity: obj.severity,
+        confirmed: obj.confirmed,
+        images: obj.images,
+        provenance: obj.provenance
+      };
+      await pubsub.publish(FAULT_CONFIRMED, payload);
+      return payload;
+    },
+
+    createPayout: async (_, { input }) => {
+      const id = uuidv4();
+      const q = `INSERT INTO payouts (id, fault_id, amount_minor_units, currency, payee_id, status)
+                 VALUES ($1,$2,$3,$4,$5,'created') RETURNING *`;
+      const vals = [id, input.faultId, input.amountMinorUnits, input.currency, input.payeeId];
+      const r = await db.query(q, vals);
+      const p = r.rows[0];
+      const payload = {
+        id: p.id,
+        faultId: p.fault_id,
+        amountMinorUnits: parseInt(p.amount_minor_units, 10),
+        currency: p.currency,
+        payeeId: p.payee_id,
+        status: p.status,
+        createdAt: p.created_at,
+        settledAt: p.settled_at,
+        txRef: p.tx_ref
+      };
+      await pubsub.publish(PAYOUT_UPDATED, payload);
+      return payload;
+    },
+
+    settlePayout: async (_, { payoutId }) => {
+      // naive settlement flow; mark processing -> settled, attach fake tx
+      const client = await db.pool.connect();
+      try {
+        await client.query('BEGIN');
+        const r = await client.query('SELECT * FROM payouts WHERE id=$1 FOR UPDATE', [payoutId]);
+        if (r.rowCount === 0) throw new Error('payout not found');
+        const payout = r.rows[0];
+        if (payout.status !== 'created' && payout.status !== 'processing') {
+          await client.query('ROLLBACK');
+          return {
+            id: payout.id,
+            faultId: payout.fault_id,
+            amountMinorUnits: parseInt(payout.amount_minor_units, 10),
+            currency: payout.currency,
+            payeeId: payout.payee_id,
+            status: payout.status,
+            createdAt: payout.created_at,
+            settledAt: payout.settled_at,
+            txRef: payout.tx_ref
+          };
+        }
+        await client.query('UPDATE payouts SET status=$1 WHERE id=$2', ['processing', payoutId]);
+        // call external payment gateway / TMS here — replaced by fake
+        const fakeTx = `TX-${Date.now()}-${Math.floor(Math.random()*1000)}`;
+        await client.query('UPDATE payouts SET status=$1, tx_ref=$2, settled_at=now() WHERE id=$3', ['settled', fakeTx, payoutId]);
+        await client.query('COMMIT');
+
+        const updated = (await db.query('SELECT * FROM payouts WHERE id=$1', [payoutId])).rows[0];
+        const payload = {
+          id: updated.id,
+          faultId: updated.fault_id,
+          amountMinorUnits: parseInt(updated.amount_minor_units, 10),
+          currency: updated.currency,
+          payeeId: updated.payee_id,
+          status: updated.status,
+          createdAt: updated.created_at,
+          settledAt: updated.settled_at,
+          txRef: updated.tx_ref
+        };
+        await pubsub.publish(PAYOUT_UPDATED, payload);
+        return payload;
+      } catch (err) {
+        await client.query('ROLLBACK');
+        throw err;
+      } finally {
+        client.release();
+      }
+    }
+  },
+
+  Subscription: {
+    faultCreated: { subscribe: () => pubsub.asyncIterator([FAULT_CREATED]) },
+    faultConfirmed: { subscribe: () => pubsub.asyncIterator([FAULT_CONFIRMED]) },
+    payoutUpdated: { subscribe: () => pubsub.asyncIterator([PAYOUT_UPDATED]) }
+  }
+};
+
+module.exports = resolvers;
+
+// -------------------------
+// src/index.js
+// -------------------------
+
+/* src/index.js */
+const { ApolloServer } = require('apollo-server');
+const typeDefs = require('./schema');
+const resolvers = require('./resolvers');
+require('dotenv').config();
+
+async function start() {
+  const server = new ApolloServer({ typeDefs, resolvers });
+  const { url } = await server.listen({ port: process.env.PORT || 4000 });
+  console.log(`🚀 Server ready at ${url}`);
+}
+
+start();
+
+// -------------------------
+// README.md
+// -------------------------
+
+/* README.md
+# Integral Apollo Server (MVP)
+
+This repo is a minimal Apollo GraphQL server wired to Postgres + Redis as Pub/Sub.
+It implements the "self-pay network" flows: ingest fault → confirm → create payout → settle payout.
+
+## Quickstart (Docker)
+1. Copy `.env.example` to `.env` and adjust if needed.
+2. Start services:
+
+   docker-compose up -d
+
+3. Run migrations into Postgres (example):
+
+   docker exec -i $(docker ps -q -f name=postgres) psql -U integral -d integral -f migrations/01_schema.sql
+
+4. Install dependencies and start app locally (or use the provided app container):
+
+   npm install
+   npm run dev
+
+5. Open GraphQL playground at http://localhost:4000/
+
+## Basic GraphQL examples
+- Ingest fault (mutation) — will publish `faultCreated` subscription
+
+mutation Ingest {
+  ingestFault(input:{ namespace: infrastructure, type: "pothole-detection", timestamp: "2025-11-30T18:00:00Z", severity: 2, location: {"lat": -29.12, "lon": 26.22} }) {
+    id
+    severity
+    confirmed
+  }
+}
+
+- Confirm fault (mutation) — will publish `faultConfirmed` and auto-create payout (in your custom flow if you wire it)
+
+mutation Confirm { confirmFault(id: "<id>", confirmed: true) { id confirmed } }
+
+- Create payout
+
+mutation CreatePayout { createPayout(input:{ faultId: "<id>", amountMinorUnits: 5000, currency: "ZAR", payeeId: "local-contractor-123" }) { id status }
+}
+
+- Settle payout
+
+mutation Settle { settlePayout(payoutId: "<payoutId>") { id status txRef }
+}
+
+## Next steps
+- Replace fake logic with ML verification service (image segmentation + sensor fusion)
+- Integrate real TMS / payment gateway in `settlePayout`
+- Add auth & RBAC for protected mutations
+- Add audit trail table and signed provenance verification
+*/