114 lines
3.7 KiB
TypeScript
114 lines
3.7 KiB
TypeScript
import { EventEmitter } from 'node:events';
|
|
import { setTimeout } from 'timers/promises';
|
|
|
|
import type Log from 'bunyan';
|
|
import type { Want, Offer } from '@prisma/client';
|
|
|
|
import { Pg } from './pg';
|
|
import { Rpc } from './rpc';
|
|
import { wantDetails } from './utils';
|
|
import routerAuth from './routers/router.auth';
|
|
|
|
export enum Events {
|
|
WantsAdded = 'WantsAdded',
|
|
Sync = 'Sync',
|
|
Connection = 'Connection',
|
|
|
|
ListingsUpdated = 'ListingsUpdated',
|
|
OffersUpdated = 'OffersUpdated',
|
|
WantsUpdated = 'WantsUpdated',
|
|
}
|
|
|
|
export default function createEvents(log: Log, pg: Pg, discogs, offers) {
|
|
const events = new EventEmitter({ captureRejections: true });
|
|
const connections = new Map<UUID, Rpc>();
|
|
|
|
events.on('error', function eventError(e) {
|
|
log.error('event error', e);
|
|
});
|
|
|
|
async function populateListings(want: Want) {
|
|
try {
|
|
const listings = await discogs.listings(want);
|
|
await pg.wantSynced(want);
|
|
return listings;
|
|
} catch (e) {
|
|
if (e.message.includes('rate limited')) {
|
|
log.warn('rate limited, trying again in 60s');
|
|
await setTimeout(60 * 1000);
|
|
return populateListings(want);
|
|
}
|
|
throw e;
|
|
}
|
|
}
|
|
|
|
async function updateWantListings(wants: Want[]) {
|
|
for (const want of wants) {
|
|
const listings = await populateListings(want);
|
|
log.info({ listings: listings.length, ...wantDetails(want) }, 'listings found');
|
|
await offers.generate();
|
|
await notifyUser(want.userId);
|
|
}
|
|
|
|
log.info({ wants: wants.length }, 'finished syncing');
|
|
}
|
|
|
|
async function notifyUser(id: UUID) {
|
|
const rpc = connections.get(id);
|
|
if (!rpc) return null;
|
|
rpc.pushUserWants();
|
|
rpc.pushUserOffers();
|
|
};
|
|
|
|
events.on(Events.WantsAdded, async function wantsAdded(wants: Want[]) {
|
|
log.info({ event: Events.WantsAdded, wants: wants.map(w => wantDetails(w)) }, 'starting listings population');
|
|
await updateWantListings(wants);
|
|
log.info({ event: Events.WantsAdded, wants: wants.map(w => wantDetails(w)) }, 'finished populating listings');
|
|
});
|
|
|
|
events.on(Events.Sync, async function sync() {
|
|
try {
|
|
const wants = await pg.wantsToSync();
|
|
log.info({ event: Events.Sync, wants: wants.length });
|
|
await updateWantListings(wants);
|
|
|
|
if (wants.length === 0) {
|
|
log.info('all listings up to date');
|
|
await setTimeout(60 * 1000)
|
|
}
|
|
} catch (e) {
|
|
log.error(e, 'sync error')
|
|
}
|
|
|
|
// forever
|
|
events.emit(Events.Sync);
|
|
});
|
|
|
|
// events.on(Events.ListingsUpdated, async function listingsUpdated() {
|
|
// const updatedOffers = await offers.generate()
|
|
// log.debug({ offers: updatedOffers.length }, 'offers up to date');
|
|
// });
|
|
|
|
events.on(Events.Connection, async function onConnection({ rpc, id }) {
|
|
log.debug('new rpc connection', { id, rpc });
|
|
connections.set(id, rpc);
|
|
})
|
|
|
|
// emitted by listen/notify
|
|
// disabled for now
|
|
// events.on(Events.OffersUpdated, async function onOffersUpdated(offer: Offer) {
|
|
// const rpc = connections.get(offer.userId);
|
|
// if (!rpc) return null;
|
|
// log.debug({ event: Events.OffersUpdated, rpc, offer });
|
|
// return rpc.pushUserOffers();
|
|
// });
|
|
|
|
// events.on(Events.WantsUpdated, async function onWantsUpdated(want: Want) {
|
|
// const rpc = connections.get(want.userId);
|
|
// if (!rpc) return null;
|
|
// log.debug({ event: Events.WantsUpdated, rpc, want });
|
|
// return rpc.pushUserWants();
|
|
// });
|
|
|
|
return events;
|
|
} |