-
Notifications
You must be signed in to change notification settings - Fork 15
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
- Loading branch information
Showing
19 changed files
with
559 additions
and
19 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,18 @@ | ||
| import _ from "lodash"; | ||
| import cron from "node-cron"; | ||
|
|
||
| import * as realtimeQueueListings from "./queues/realtime-queue-listings"; | ||
| import { acquireLock } from "../../common/redis"; | ||
| import { config } from "../../config"; | ||
| import { Okx } from "../../utils/okx"; | ||
|
|
||
| if (config.doRealtimeWork) { | ||
| if (new Okx().getChainName()) { | ||
| cron.schedule("*/20 * * * * *", async () => { | ||
| const lockAcquired = await acquireLock(realtimeQueueListings.getLockKey(), 30); | ||
| if (lockAcquired) { | ||
| await realtimeQueueListings.addToRealtimeQueue(); | ||
| } | ||
| }); | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,77 @@ | ||
| import { Job, Queue, QueueScheduler, Worker } from "bullmq"; | ||
|
|
||
| import { logger } from "../../../common/logger"; | ||
| import { redis } from "../../../common/redis"; | ||
| import { config } from "../../../config"; | ||
| import { fetchOrders } from "../utils"; | ||
|
|
||
| const BACKFILL_QUEUE_NAME = "backfill-okx-listings-sync"; | ||
|
|
||
| export const backfillQueue = new Queue(BACKFILL_QUEUE_NAME, { | ||
| connection: redis.duplicate(), | ||
| defaultJobOptions: { | ||
| attempts: 1, | ||
| backoff: { | ||
| type: "fixed", | ||
| delay: 3, | ||
| }, | ||
| timeout: 60000, | ||
| removeOnComplete: 100, | ||
| removeOnFail: 1000, | ||
| }, | ||
| }); | ||
| new QueueScheduler(BACKFILL_QUEUE_NAME, { connection: redis.duplicate() }); | ||
|
|
||
| if (config.doBackfillWork) { | ||
| const realtimeWorker = new Worker( | ||
| BACKFILL_QUEUE_NAME, | ||
| async (job: Job) => { | ||
| const { runId } = job.data; | ||
|
|
||
| try { | ||
| const createBefore = await redis | ||
| .get(getCreateBeforeKey(runId)) | ||
| .then((c) => (c ? c : Math.floor(Date.now() / 1000))); | ||
|
|
||
| logger.info( | ||
| BACKFILL_QUEUE_NAME, | ||
| `Start syncing OKX listings (runId=${runId} createBefore=${createBefore})` | ||
| ); | ||
|
|
||
| const { minTimestamp } = await fetchOrders({ | ||
| side: "sell", | ||
| createBefore: Number(createBefore), | ||
| maxIterations: 10, | ||
| }); | ||
| if (minTimestamp && minTimestamp + 1 !== Number(createBefore)) { | ||
| await redis.set(getCreateBeforeKey(runId), minTimestamp + 1); | ||
| await addToOkxBackfillQueue(runId); | ||
| } | ||
| } catch (error) { | ||
| logger.error( | ||
| BACKFILL_QUEUE_NAME, | ||
| JSON.stringify({ | ||
| message: `OKX listings sync failed (runId=${runId})`, | ||
| error, | ||
| stack: (error as any).stack, | ||
| attempts: job.attemptsMade, | ||
| syncSource: "OKX", | ||
| }) | ||
| ); | ||
| throw error; | ||
| } | ||
| }, | ||
| { connection: redis.duplicate(), concurrency: 1 } | ||
| ); | ||
|
|
||
| realtimeWorker.on("error", async (error) => { | ||
| logger.error(BACKFILL_QUEUE_NAME, `Worker errored: ${error}`); | ||
| }); | ||
| } | ||
|
|
||
| export const addToOkxBackfillQueue = async (runId = "", delayMs = 0) => { | ||
| await backfillQueue.add(BACKFILL_QUEUE_NAME, { runId }, { delay: delayMs }); | ||
| }; | ||
|
|
||
| const getCreateBeforeKey = (runId: string) => `${BACKFILL_QUEUE_NAME}-${runId}-create-before`; | ||
| export const getLockKey = (runId: string) => `${BACKFILL_QUEUE_NAME}-${runId}-lock`; |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,83 @@ | ||
| import { Job, Queue, QueueScheduler, Worker } from "bullmq"; | ||
|
|
||
| import { logger } from "../../../common/logger"; | ||
| import { redis, releaseLock } from "../../../common/redis"; | ||
| import { config } from "../../../config"; | ||
| import { fetchOrders } from "../utils"; | ||
|
|
||
| const REALTIME_QUEUE_NAME = "realtime-okx-listings-sync"; | ||
|
|
||
| export const realtimeQueue = new Queue(REALTIME_QUEUE_NAME, { | ||
| connection: redis.duplicate(), | ||
| defaultJobOptions: { | ||
| attempts: 1, | ||
| backoff: { | ||
| type: "fixed", | ||
| delay: 3, | ||
| }, | ||
| timeout: 60000, | ||
| removeOnComplete: 100, | ||
| removeOnFail: 1000, | ||
| }, | ||
| }); | ||
| new QueueScheduler(REALTIME_QUEUE_NAME, { connection: redis.duplicate() }); | ||
|
|
||
| if (config.doRealtimeWork) { | ||
| const realtimeWorker = new Worker( | ||
| REALTIME_QUEUE_NAME, | ||
| async (job: Job) => { | ||
| try { | ||
| const createAfter = await redis | ||
| .get(getCreateAfterKey()) | ||
| .then((c) => (c ? c : Math.floor(Date.now() / 1000 - 30))); | ||
|
|
||
| logger.info(REALTIME_QUEUE_NAME, `Start syncing OKX listings (createAfter=${createAfter})`); | ||
|
|
||
| const { maxTimestamp } = await fetchOrders({ | ||
| side: "sell", | ||
| createAfter: Number(createAfter), | ||
| }); | ||
| if (maxTimestamp) { | ||
| await redis.set(getCreateAfterKey(), maxTimestamp - 1); | ||
| } | ||
| } catch (error) { | ||
| logger.error( | ||
| REALTIME_QUEUE_NAME, | ||
| JSON.stringify({ | ||
| message: "OKX listings sync failed", | ||
| error, | ||
| stack: (error as any).stack, | ||
| attempts: job.attemptsMade, | ||
| syncSource: "OKX", | ||
| }) | ||
| ); | ||
| throw error; | ||
| } | ||
| }, | ||
| { connection: redis.duplicate(), concurrency: 2 } | ||
| ); | ||
|
|
||
| realtimeWorker.on("completed", async (job) => { | ||
| await releaseLock(getLockKey(), false); | ||
|
|
||
| if (job.attemptsMade > 0) { | ||
| logger.info( | ||
| REALTIME_QUEUE_NAME, | ||
| `OKX listings sync recovered (attempts=${job.attemptsMade})` | ||
| ); | ||
| } | ||
| }); | ||
|
|
||
| realtimeWorker.on("error", async (error) => { | ||
| await releaseLock(getLockKey(), false); | ||
|
|
||
| logger.error(REALTIME_QUEUE_NAME, `Worker errored: ${error}`); | ||
| }); | ||
| } | ||
|
|
||
| export const addToRealtimeQueue = async (delayMs: number = 0) => { | ||
| await realtimeQueue.add(REALTIME_QUEUE_NAME, {}, { delay: delayMs }); | ||
| }; | ||
|
|
||
| const getCreateAfterKey = () => `${REALTIME_QUEUE_NAME}-create-after`; | ||
| export const getLockKey = () => `${REALTIME_QUEUE_NAME}-lock`; |
Oops, something went wrong.