This commit is contained in:
ntr committed 2022-08-18 00:34:46 +10:00
1 parent 30bf7426e4
commit 323c7877b1
13 files changed
+94 -52

No files matched your search

+4
View File
@@ -1,5 +1,6 @@
import createServices from './src/services';
import createLog from './src/log';
import { Events } from './src/events';
async function start() {
const log = createLog();
@@ -14,8 +15,11 @@ async function start() {
pg,
discogs,
api,
events,
} = createServices(log);
events.emit(Events.StartupSync);
await api.listen();
}
+3 -6
View File
@@ -153,12 +153,9 @@ export default function createApi(log: Log, pg: Pg, discogs, events) {
const { releaseId } = ctx.request.body;
const user = await pg.userFindId(ctx.state.user.id);
const release = await discogs.release(releaseId);
const want = await pg.wantCreate([release], user);
// fetch listings
const listings = await discogs.listings(want);
ctx.response.body = { want, listings };
const want = await pg.wantCreate(release, user);
events.emit(Events.WantsAdded, [want]);
ctx.response.body = [want];
});
router.post('/want/discogs', async function wantListPost(ctx, next) {
+1 -1
View File
@@ -84,7 +84,7 @@ export default function discogs(log: Log, pg: Pg) {
try {
log.debug('fetching listing', id);
const res = await api.get(`${apiBase}/marketplace/listings/${id}`);
log.info(res.data);
log.debug(res.data);
return pg.listingUpsert(res.data, want);
} catch (e) {
return handleApiError(e);
+28 -3
View File
@@ -1,25 +1,50 @@
import { EventEmitter } from 'node:events';
import type Log from 'bunyan';
import type { Listing, User, Want, Prisma } from '@prisma/client';
import type { Want } from '@prisma/client';
import { Pg } from './pg';
import routerAuth from './routers/router.auth';
export enum Events {
WantsAdded = 'WantsAdded',
StartupSync = 'StartupSync',
}
export function eventService(log: Log, pg: Pg, discogs) {
const events = new EventEmitter({ captureRejections: true });
events.on('error', function eventError(e) {
log.error(e);
log.error('event error', e);
});
events.on(Events.WantsAdded, async function wantsAdded(wants: Want[]) {
log.info('wants added', wants);
for (const want of wants) {
const listings = await discogs.listings(want);
log.info({ count: listings.length, title: want.id }, 'listings added')
}
log.info('finished fetching listings', { wants: wants.length });
});
events.on(Events.StartupSync, async function wantsAdded() {
const wants = await pg.wantsToSync();
try {
for (const want of wants) {
const listings = await discogs.listings(want);
log.info({ count: listings.length, title: want.id }, 'listings added')
}
log.info('finished syncing', { wants: wants.length });
} catch (e) {
if (e.message === 'rate limited') {
log.warn('rate limited, trying again in 60s');
return setTimeout(() => events.emit(Events.StartupSync), 60 * 1000);
}
}
});
return events;
}
+6
View File
@@ -18,6 +18,7 @@ export interface Pg {
wantCreate: (discogsWant: any, user: User) => Promise<Want>,
wantList: (user: User) => Promise<Want[]>,
wantsToSync: () => Promise<Want[]>,
listingUpsert: (listing: Listing, want: Want) => Promise<Listing>,
listingFind: (id: DiscogsListing) => Promise<Listing>,
listingRmUnlisted: (want: Want, ids: DiscogsListing[]) => Promise<number>,
@@ -60,6 +61,10 @@ export default function createPg(log: Log, prisma: PrismaClient): Pg {
});
}
async function wantsToSync() {
return prisma.want.findMany({});
}
async function wantCreate(discogsRelease, user) {
const data: { userId: UUID, release: DiscogsListing, data: any } = {
data: discogsRelease,
@@ -149,6 +154,7 @@ export default function createPg(log: Log, prisma: PrismaClient): Pg {
wantCreate,
wantList,
wantsToSync,
listingUpsert,
listingFind,
listingRmUnlisted,