A social knowledge tool for researchers built on ATProto
cosmik.network
1import { IProcess } from '../../domain/IProcess';
2import { EnvironmentConfigService } from '../config/EnvironmentConfigService';
3import { RepositoryFactory } from '../http/factories/RepositoryFactory';
4import { ServiceFactory } from '../http/factories/ServiceFactory';
5import { UseCaseFactory } from '../http/factories/UseCaseFactory';
6import { CardAddedToLibraryEventHandler } from '../../../modules/feeds/application/eventHandlers/CardAddedToLibraryEventHandler';
7import { CardAddedToCollectionEventHandler } from '../../../modules/feeds/application/eventHandlers/CardAddedToCollectionEventHandler';
8import { CardCollectionSaga } from '../../../modules/feeds/application/sagas/CardCollectionSaga';
9import { QueueNames } from '../events/QueueConfig';
10import { EventNames } from '../events/EventConfig';
11
12export class FeedWorkerProcess implements IProcess {
13 constructor(private configService: EnvironmentConfigService) {}
14
15 async start(): Promise<void> {
16 console.log('Starting feed worker...');
17
18 // Create dependencies using factories
19 const repositories = RepositoryFactory.create(this.configService);
20 const services = ServiceFactory.createForWorker(
21 this.configService,
22 repositories,
23 );
24 const useCases = UseCaseFactory.createForWorker(repositories, services);
25
26 // Test Redis connection (only if using Redis)
27 if (services.redisConnection) {
28 try {
29 await services.redisConnection.ping();
30 console.log('Connected to Redis successfully');
31 } catch (error) {
32 console.error('Failed to connect to Redis:', error);
33 process.exit(1);
34 }
35 } else {
36 console.log('Using in-memory event system');
37 }
38
39 // Create subscriber for feeds queue
40 const eventSubscriber = services.createEventSubscriber(QueueNames.FEEDS);
41
42 // Create the saga with proper dependencies
43 const cardCollectionSaga = new CardCollectionSaga(
44 useCases.addActivityToFeedUseCase,
45 );
46
47 // Create event handlers with the saga
48 const cardAddedToLibraryHandler = new CardAddedToLibraryEventHandler(
49 cardCollectionSaga,
50 );
51 const cardAddedToCollectionHandler = new CardAddedToCollectionEventHandler(
52 cardCollectionSaga,
53 );
54
55 // Register handlers
56 await eventSubscriber.subscribe(
57 EventNames.CARD_ADDED_TO_LIBRARY,
58 cardAddedToLibraryHandler,
59 );
60
61 await eventSubscriber.subscribe(
62 EventNames.CARD_ADDED_TO_COLLECTION,
63 cardAddedToCollectionHandler,
64 );
65
66 // Start the worker
67 await eventSubscriber.start();
68
69 console.log('Feed worker started and listening for events...');
70
71 // Graceful shutdown
72 const shutdown = async () => {
73 console.log('Shutting down feed worker...');
74 await eventSubscriber.stop();
75 if (services.redisConnection) {
76 await services.redisConnection.quit();
77 }
78 process.exit(0);
79 };
80
81 process.on('SIGTERM', shutdown);
82 process.on('SIGINT', shutdown);
83 }
84}