Add SinkingYachts realtime feed integration

package.json

@@ -47,7 +47,8 @@
     "request-ip": "^3.3.0",
     "resend": "^4.2.0",
     "response-time": "^2.3.3",
-    "ts-node": "^10.9.2"
+    "ts-node": "^10.9.2",
+    "ws": "^8.18.0"
   },
   "devDependencies": {
     "@types/react": "^19.0.12",
@@ -60,6 +61,7 @@
     "@types/node-statsd": "^0.1.6",
     "@types/request-ip": "^0.0.41",
     "@types/response-time": "^2.3.8",
+    "@types/ws": "^8.5.13",
     "nodemon": "^3.1.9",
     "typescript": "^5.7.2"
   },

src/admin-routes/router.ts

@@ -14,6 +14,7 @@ import { logRequest } from "src/middleware/logRequest";
 import { db } from "src/utils/db";
 import { authenticateToken, getUserInfo } from "src/utils/jwt";
 import domainRouter from "./routes/domain";
+import sinkingYachtsRouter from "./routes/sinking-yachts";
 import userRouter from "./routes/user";
 const router = express.Router();
 router.use(express.json());
@@ -297,6 +298,7 @@ router.get("/metrics", logRequest, async (req, res) => {
 });
 
 router.use("/domain", logRequest, domainRouter);
+router.use("/sinking-yachts", logRequest, sinkingYachtsRouter);
 router.use("/user", userRouter);
 
 export default router;

src/admin-routes/routes/sinking-yachts.ts

@@ -0,0 +1,123 @@
+import express, { Request, Response } from "express";
+import { sinkingYahtsService } from "src/services/_index";
+
+const router = express.Router();
+
+/**
+ * POST /admin/sinking-yachts/start-feed
+ * @summary Starts the SinkingYachts realtime feed monitoring
+ * @tags SinkingYachts - Feed Management
+ * @security BearerAuth
+ * @param {object} request.body
+ * @param {boolean} request.body.skipBulkImport - Skip the initial bulk import (optional)
+ * @return {object} 200 - Success response
+ * @return {object} 500 - Error response
+ */
+router.post("/start-feed", async (req: Request, res: Response) => {
+  try {
+    const { skipBulkImport = false } = req.body;
+
+    await sinkingYahtsService.startFeedMonitoring(skipBulkImport);
+
+    return res.status(200).json({
+      success: true,
+      message: "SinkingYachts feed monitoring started successfully",
+      skipBulkImport
+    });
+  } catch (error: any) {
+    return res.status(500).json({
+      success: false,
+      message: "Failed to start SinkingYachts feed monitoring",
+      error: error.message
+    });
+  }
+});
+
+/**
+ * POST /admin/sinking-yachts/stop-feed
+ * @summary Stops the SinkingYachts realtime feed monitoring
+ * @tags SinkingYachts - Feed Management  
+ * @security BearerAuth
+ * @return {object} 200 - Success response
+ */
+router.post("/stop-feed", async (req: Request, res: Response) => {
+  try {
+    sinkingYahtsService.stopRealtimeFeed();
+
+    return res.status(200).json({
+      success: true,
+      message: "SinkingYachts feed monitoring stopped successfully"
+    });
+  } catch (error: any) {
+    return res.status(500).json({
+      success: false,
+      message: "Failed to stop SinkingYachts feed monitoring",
+      error: error.message
+    });
+  }
+});
+
+/**
+ * POST /admin/sinking-yachts/bulk-import
+ * @summary Manually triggers a bulk import of all domains from SinkingYachts
+ * @tags SinkingYachts - Feed Management
+ * @security BearerAuth
+ * @return {object} 200 - Success response
+ * @return {object} 500 - Error response
+ */
+router.post("/bulk-import", async (req: Request, res: Response) => {
+  try {
+    await sinkingYahtsService.initializeBulkImport();
+
+    return res.status(200).json({
+      success: true,
+      message: "SinkingYachts bulk import completed successfully"
+    });
+  } catch (error: any) {
+    return res.status(500).json({
+      success: false,
+      message: "Failed to complete SinkingYachts bulk import",
+      error: error.message
+    });
+  }
+});
+
+/**
+ * GET /admin/sinking-yachts/recent
+ * @summary Fetches recent domains from SinkingYachts API
+ * @tags SinkingYachts - Feed Management
+ * @security BearerAuth
+ * @param {string} since.query.required - ISO date string for filtering recent domains
+ * @return {object} 200 - Array of recent domains
+ * @return {object} 400 - Bad request (missing since parameter)
+ * @return {object} 500 - Error response
+ */
+router.get("/recent", async (req: Request, res: Response) => {
+  try {
+    const { since } = req.query;
+
+    if (!since || typeof since !== "string") {
+      return res.status(400).json({
+        success: false,
+        message: "Missing or invalid 'since' query parameter (ISO date string required)"
+      });
+    }
+
+    const recentDomains = await sinkingYahtsService.getRecentDomains(since);
+
+    return res.status(200).json({
+      success: true,
+      data: recentDomains,
+      count: recentDomains.length,
+      since
+    });
+  } catch (error: any) {
+    return res.status(500).json({
+      success: false,
+      message: "Failed to fetch recent domains",
+      error: error.message
+    });
+  }
+});
+
+export default router;

src/index.ts

@@ -11,6 +11,7 @@ import { server } from "./server";
 // import metrics from "./metrics";
 import { swaggerOptions as adminSwagOptions } from "./admin-routes/swaggerOptions";
 import { swaggerOptions as mainSwagOptions } from "./swaggerOptions";
+import { sinkingYahtsService } from "./services/_index";
 import * as logger from "./utils/logger";
 dotenv.config();
 
@@ -26,6 +27,21 @@ try {
   }
   const db = drizzle(process.env.DATABASE_URL);
   logger.database("Database connection initialized successfully");
+
+  // Initialize SinkingYachts realtime feed monitoring
+  if (process.env.ENABLE_SINKING_YACHTS_FEED !== "false") {
+    logger.info("Starting SinkingYachts realtime feed monitoring...");
+    sinkingYahtsService.startFeedMonitoring(process.env.SKIP_BULK_IMPORT === "true")
+      .then(() => {
+        logger.info("SinkingYachts feed monitoring started successfully");
+      })
+      .catch((error) => {
+        logger.error(`Failed to start SinkingYachts feed monitoring: ${error.message}`);
+        // Don't exit the process, let the server continue without feed monitoring
+      });
+  } else {
+    logger.info("SinkingYachts feed monitoring disabled by ENABLE_SINKING_YACHTS_FEED=false");
+  }
 } catch (error) {
   logger.error(
     `Failed to initialize database connection: ${error instanceof Error ? error.message : String(error)}`