Eventos periodicos configurables

package.json

@@ -6,7 +6,7 @@
     "scripts": {
         "test": "vitest watch",
         "build": "npx tsc",
-        "dev": "tsx --watch ./src/server.ts",
+        "dev": "tsx --watch ./src/app.ts",
         "start": "node dist/index.js",
         "typecheck": "npx tsc --noEmit",
         "lint": "eslint .",

src/app.ts

@@ -1,27 +1,19 @@
-import express from 'express';
+import server from './server';
-import cors from 'cors';
+import { clientConfig, config } from './config/index';
-import webhookRoutes from './routes/webhook';
+import client from './client'
 
-const app = express();
+const PORT = config.port;
+const CLIENT_PORT = clientConfig.port;
 
-// Middleware
+/**
-app.use(cors());
+* TODO: Meter opcion de no lanzar el cliente
+*/
 
-// Webhooks often require raw body for signature verification if not handled by express.json()
+server.listen(PORT, () => {
-// using verify option in body-parser/express.json to get raw body if needed.
+    console.log(`[Server] Server is running on port ${PORT} in ${config.env} mode`);
-app.use(express.json({
+    console.log(`[Server] Webhook endpoint: http://localhost:${PORT}/api/webhooks`);
-    verify: (req: any, res, buf) => {
-        req.rawBody = buf;
-    }
-}));
-app.use(express.urlencoded({ extended: true }));
-
-// Webhooks
-app.use('/webhooks', webhookRoutes);
-
-// Health check
-app.get('/health', (req, res) => {
-    res.status(200).send({ resp: 'OK' });
 });
 
-export default app;
+client.listen(CLIENT_PORT, () => {
+    console.log(` [Client] Client ins running on port ${CLIENT_PORT}`)
+})

src/config/initial_hosts.ts

@@ -2,20 +2,10 @@
  * Configuracion de los host iniciales y a que topics escuchan
  */
 
-import { SubscriptionData, SubscriptorData } from "#controllers/subscriptions.js"
+import { SetupSubscription, SubscriptionData } from "#controllers/subscriptions.types.js"
 import { webhooksConfig } from "."
 
 
-type SetupSubscription = SubscriptorData & {
-    topic: string,
-    options: {
-        /* En ms. Cada cuanto se envia un mensaje nuevo, para solo se envia un mensaje */
-        period?: number,
-        /* Numero de mensajes, para undefiend solo se manda uno, para <= 0 seran infinitos */
-        mesages?: number,
-    }
-}
-
 export function generateSubscriptions(initialHosts?: SetupSubscription[]) {
     const initialhs = initialHosts || INITIAL_HOSTS
     const topicSubscriberMap = new Map<string, SubscriptionData>()

src/controllers/subscription_manager.ts

@@ -0,0 +1,102 @@
+import { generateHMACSignature } from "#middleware/hmac.js";
+import { requestBuilder, shopifyHeaderBuilder } from "#shared/requests.js";
+import { ShopifyEvent, SubscriptionData, SubscriptorData, Topic } from "./subscriptions.types";
+
+/**
+ * De scheduler tiene poco
+ */
+export class EventScheduler {
+    public eventList: SubscriptorData[] = []
+    public activeIntervals: NodeJS.Timeout[] = []
+
+    constructor(args: {
+        events?: SubscriptorData[],
+    }) {
+        if (args.events != undefined) {
+            this.eventList = args.events
+        }
+        this.start()
+    }
+
+    private start() {
+        for (const event of this.eventList) {
+            let sentMesages = 0
+            const interval = setInterval(() => {
+                console.log("[Server] Lanzado evento ", event)
+                if (event.options?.mesages == undefined || event.options?.mesages < 1)
+                    return; // Se lannza de continuo
+                sentMesages++;
+                if (sentMesages > event.options.mesages) {
+                    clearInterval(interval)
+                }
+            }, event.options?.period ?? 1000)
+            this.activeIntervals.push(interval)
+        }
+    }
+
+}
+
+export class SubscriptionManager {
+    private subscriptions: Map<Topic, SubscriptionData> = new Map<Topic, SubscriptionData>()
+    public scheduler: EventScheduler
+
+    constructor(
+        scheduler: EventScheduler,
+        subs?: typeof this.subscriptions,
+    ) {
+        this.scheduler = scheduler
+        if (subs != undefined) {
+            this.subscriptions = subs
+        }
+    }
+
+
+    public addSubscriber(args: { topic: string, subscriber: SubscriptorData }) {
+        const topic = args.topic
+        if (topic == undefined) throw new Error("Topic vacio")
+        const topicSubscribers = this.subscriptions.get(topic)
+        if (topicSubscribers == undefined) {
+            this.subscriptions.set(topic, {
+                topic,
+                subscriptors: [args.subscriber]
+            })
+        }
+    }
+
+    public poll(event: ShopifyEvent) {
+        const topic = event.topic
+
+        if (!this.subscriptions.has(topic)) {
+            console.error("Topic desconocido: " + topic)
+            return;
+        }
+
+        const subscriptors = this.subscriptions.get(topic)
+
+        for (const sub of subscriptors!.subscriptors) {
+            const body = {
+                id: 1234
+            }
+            const parsedBody = JSON.stringify(body)
+            const signature = generateHMACSignature(parsedBody, sub.secretkey)
+            const request = requestBuilder({
+                method: sub.method,
+                headers: shopifyHeaderBuilder({
+                    topic: topic,
+                    signature: signature
+                }),
+                host: sub.host,
+                port: sub.port,
+                endpoint: sub.endpoint
+            })
+            request.write(parsedBody)
+            // Data puede venir en chunks!
+            request.on("data", () => console.log)
+            request.on("end", () => console.log)
+            request.on("error", () => console.error)
+
+            request.end()
+            console.debug("Enviado evento a ", sub.host, ":", sub.port, sub.endpoint)
+        }
+    }
+}

src/controllers/subscriptions.ts

@@ -2,32 +2,10 @@
  * Simulacion de que clientes estan subscritos a que eventos
  * */
 
-import { Shopify } from "../data/webhooks/order"
-import http from "node:http"
 import { Request, Response } from 'express';
-import { generateHMACSignature } from "#middleware/hmac.js"
 import { config, webhooksConfig } from "#config/index.js";
-
+import { SubscriptionData, Topic, SubscriptorData, ShopifyEvent } from "./subscriptions.types"
-export type SubscriptionData = {
+import { SubscriptionManager } from './subscription_manager';
-    topic: string,
-    subscriptors: SubscriptorData[]
-}
-
-export type SubscriptorData = {
-    host: string,
-    port: string,
-    endpoint: string,
-    secretkey: string,
-    method: "POST" | "GET" | "PUT" | "DELETE"
-    open: Date,
-}
-
-export type ShopifyEvent = {
-    topic: string,
-    data: Shopify.OrderCreate | Shopify.OrderUpdate | Shopify.OrderDelete
-}
-
-type Topic = string
 
 /** Mapa topic -> subscriber */
 export const subscribers = new Map<Topic, SubscriptionData>()
@@ -63,105 +41,3 @@ export function subscriptonHandlerBuilder(subscriptionManager: SubscriptionManag
         }
     }
 }
-
-export class SubscriptionManager {
-    private subscriptions: Map<Topic, SubscriptionData> = new Map<Topic, SubscriptionData>()
-
-    constructor(subs?: typeof this.subscriptions) {
-        if (subs != undefined) {
-            this.subscriptions = subs
-        }
-    }
-
-    public addSubscriber(args: { topic: string, subscriber: SubscriptorData }) {
-        const topic = args.topic
-        if (topic == undefined) throw new Error("Topic vacio")
-        const topicSubscribers = this.subscriptions.get(topic)
-        if (topicSubscribers == undefined) {
-            this.subscriptions.set(topic, {
-                topic,
-                subscriptors: [args.subscriber]
-            })
-        }
-    }
-
-    public poll(event: ShopifyEvent) {
-        const topic = event.topic
-
-        if (!this.subscriptions.has(topic)) {
-            console.error("Topic desconocido: " + topic)
-            return;
-        }
-
-        const subscriptors = this.subscriptions.get(topic)
-
-        for (const sub of subscriptors!.subscriptors) {
-            const body = {
-                id: 1234
-            }
-            const parsedBody = JSON.stringify(body)
-            const signature = generateHMACSignature(parsedBody, sub.secretkey)
-            const request = requestBuilder({
-                method: sub.method,
-                headers: shopifyHeaderBuilder({
-                    topic: topic,
-                    signature: signature
-                }),
-                host: sub.host,
-                port: sub.port,
-                endpoint: sub.endpoint
-            })
-            request.write(parsedBody)
-            // Data puede venir en chunks!
-            request.on("data", () => console.log)
-            request.on("end", () => console.log)
-            request.on("error", () => console.error)
-
-            request.end()
-            console.debug("Enviado evento a ", sub.host, ":", sub.port, sub.endpoint)
-        }
-    }
-}
-
-// TODO: Crear una fucnion especifica para los request con HMAC
-function requestBuilder(args: {
-    method: string,
-    headers: Object,
-    host: string,
-    port: string,
-    endpoint: string,
-}) {
-    const request = http.request({
-        host: args.host,
-        port: args.port,
-        path: args.endpoint
-    })
-
-    Object.entries(args.headers).forEach(([name, value]) => {
-        request.setHeader(name, String(value))
-    })
-    return request
-}
-
-function headerBuilder(args: {}) {
-    return {
-        "host": "127.0.0.1",
-        "Content-Type": "application/json"
-    }
-}
-
-/**
-* TODO: Estoy confiando que la firma esté ok 
-*/
-function shopifyHeaderBuilder(args: {
-    topic: string,
-    signature: string
-}) {
-    return {
-        ...headerBuilder(args),
-        "x-shopify-topic": args.topic,
-        "x-shopify-hmac-sha256": args.signature
-    }
-}
-
-

src/controllers/subscriptions.types.ts

@@ -0,0 +1,35 @@
+import { Shopify } from "../data/webhooks/order"
+
+export type SubscriptionData = {
+    topic: string,
+    subscriptors: SubscriptorData[]
+}
+
+export type SubscriptorData = {
+    host: string,
+    port: string,
+    endpoint: string,
+    secretkey: string,
+    method: "POST" | "GET" | "PUT" | "DELETE"
+    open: Date,
+    options?: {
+        /* En ms. Cada cuanto se envia un mensaje nuevo, para solo se envia un mensaje */
+        period?: number,
+        /* Numero de mensajes, para undefiend solo se manda uno, para <= 0 seran infinitos */
+        mesages?: number,
+    }
+}
+
+export type ShopifyEvent = {
+    topic: string,
+    data: Shopify.OrderCreate | Shopify.OrderUpdate | Shopify.OrderDelete
+}
+
+export type Topic = string
+
+/**
+ * Solo para inicializar las subscripciones
+ */
+export type SetupSubscription = SubscriptorData & {
+    topic: string
+}

src/routes/webhook.ts

@@ -1,12 +1,15 @@
 import { Router } from 'express';
 import { ordersHandlerBuilder } from '#controllers/orders.webhook.js';
-import { SubscriptionManager, subscriptonHandlerBuilder } from '#controllers/subscriptions.js';
+import { subscriptonHandlerBuilder } from '#controllers/subscriptions.js';
 import { generateSubscriptions } from '#config/initial_hosts.js';
+import { EventScheduler, SubscriptionManager } from '#controllers/subscription_manager.js';
 
 const webhookRouter = Router();
 
 const baseSubscriptions = generateSubscriptions()
-const subscriptions = new SubscriptionManager(baseSubscriptions)
+const subscriptionList = [...baseSubscriptions.values()].map(e => e.subscriptors).flat()
+const scheduler = new EventScheduler({ events: subscriptionList })
+const subscriptions = new SubscriptionManager(scheduler, baseSubscriptions)
 
 // subto
 webhookRouter.post('/subto', subscriptonHandlerBuilder(subscriptions));

src/server.ts

@@ -1,16 +1,27 @@
-import app from './app';
+import express from 'express';
-import { clientConfig, config } from './config/index';
+import cors from 'cors';
-import client from './client'
+import webhookRoutes from './routes/webhook';
 
-const PORT = config.port;
+const server = express();
-const CLIENT_PORT = clientConfig.port;
 
-app.listen(PORT, () => {
+// Middleware
-    console.log(`Server is running on port ${PORT} in ${config.env} mode`);
+server.use(cors());
-    console.log(`Webhook endpoint: http://localhost:${PORT}/api/webhooks`);
+
+// Webhooks often require raw body for signature verification if not handled by express.json()
+// using verify option in body-parser/express.json to get raw body if needed.
+server.use(express.json({
+    verify: (req: any, res, buf) => {
+        req.rawBody = buf;
+    }
+}));
+server.use(express.urlencoded({ extended: true }));
+
+// Webhooks
+server.use('/webhooks', webhookRoutes);
+
+// Health check
+server.get('/health', (req, res) => {
+    res.status(200).send({ resp: 'OK' });
 });
 
-
+export default server;
-client.listen(CLIENT_PORT, () => {
-    console.log(`Client ins running on port ${CLIENT_PORT}`)
-})

src/shared/requests.ts

@@ -0,0 +1,44 @@
+import http from "node:http"
+
+// TODO: Crear una fucnion especifica para los request con HMAC
+export function requestBuilder(args: {
+    method: string,
+    headers: Object,
+    host: string,
+    port: string,
+    endpoint: string,
+}) {
+    const request = http.request({
+        host: args.host,
+        port: args.port,
+        path: args.endpoint
+    })
+
+    Object.entries(args.headers).forEach(([name, value]) => {
+        request.setHeader(name, String(value))
+    })
+    return request
+}
+
+export function headerBuilder(args: {}) {
+    return {
+        "host": "127.0.0.1",
+        "Content-Type": "application/json"
+    }
+}
+
+/**
+* TODO: Estoy confiando que la firma esté ok 
+*/
+export function shopifyHeaderBuilder(args: {
+    topic: string,
+    signature: string
+}) {
+    return {
+        ...headerBuilder(args),
+        "x-shopify-topic": args.topic,
+        "x-shopify-hmac-sha256": args.signature
+    }
+}
+
+

src/tests/setup.ts

@@ -2,7 +2,7 @@
 import { Server } from "node:http";
 import { beforeAll, afterAll } from "vitest";
 import { config } from '#config/index.js';
-import app from "#app.js"
+import server from "#server.js"
 
 import express from 'express';
 import supertest, { Test } from "supertest";
@@ -13,7 +13,7 @@ let request: TestAgent<Test>;
 
 beforeAll(() => {
     const PORT = config.port;
-    testServer = app.listen(PORT, (err) => {
+    testServer = server.listen(PORT, (err) => {
         if (err != undefined) {
             console.error(err)
         } else {
@@ -22,7 +22,7 @@ beforeAll(() => {
         }
     });
     console.log("! Server iniciado")
-    request = supertest(app)
+    request = supertest(server)
 })
 
 afterAll(() => {