add support for bull with redis

backend/package.json

@@ -26,6 +26,7 @@
     "@fastify/multipart": "^7.2.0",
     "@fastify/reply-from": "^8.3.0",
     "@fastify/static": "^6.5.0",
+    "@nestjs/bull": "^0.6.1",
     "@nestjs/common": "^9.0.11",
     "@nestjs/config": "^2.2.0",
     "@nestjs/core": "^9.0.11",
@@ -40,6 +41,7 @@
     "@nestjs/websockets": "^9.0.11",
     "bcrypt": "^5.0.1",
     "bmp-img": "^1.2.1",
+    "bull": "^4.9.0",
     "cors": "^2.8.5",
     "file-type": "^18.0.0",
     "is-docker": "^3.0.0",
@@ -69,6 +71,7 @@
     "@nestjs/schematics": "^9.0.3",
     "@nestjs/testing": "^9.0.11",
     "@types/bcrypt": "^5.0.0",
+    "@types/bull": "^3.15.9",
     "@types/cors": "^2.8.12",
     "@types/multer": "^1.4.7",
     "@types/node": "^18.7.18",

backend/src/app.module.ts

@@ -1,10 +1,13 @@
+import { BullModule } from '@nestjs/bull';
 import { MiddlewareConsumer, Module, NestModule } from '@nestjs/common';
 import { ScheduleModule } from '@nestjs/schedule';
 import { ServeStaticModule } from '@nestjs/serve-static';
 import cors from 'cors';
 import { IncomingMessage, ServerResponse } from 'http';
+import { BullConfigService } from './config/early/bull.config.service';
 import { EarlyConfigModule } from './config/early/early-config.module';
 import { ServeStaticConfigService } from './config/early/serve-static.config.service';
+import { ConsumersModule } from './consumers/consumers.module';
 import { DatabaseModule } from './database/database.module';
 import { PicsurLayersModule } from './layers/PicsurLayers.module';
 import { PicsurLoggerModule } from './logger/logger.module';
@@ -45,12 +48,18 @@ const imageCorsOverride = (
       imports: [EarlyConfigModule],
     }),
     ScheduleModule.forRoot(),
+    BullModule.forRootAsync({
+      useExisting: BullConfigService,
+      imports: [EarlyConfigModule],
+    }),
+
     DatabaseModule,
     AuthManagerModule,
     UsageManagerModule,
     DemoManagerModule,
     PicsurRoutesModule,
     PicsurLayersModule,
+    ConsumersModule,
   ],
 })
 export class AppModule implements NestModule {

backend/src/collections/ingress-file-db/ingress-file-db.module.ts

@@ -0,0 +1,11 @@
+import { Module } from '@nestjs/common';
+import { TypeOrmModule } from '@nestjs/typeorm';
+import { EIngressFileBackend } from '../../database/entities/ingress-file.entity';
+import { IngressFileDbService } from './ingress-file-db.service';
+
+@Module({
+  imports: [TypeOrmModule.forFeature([EIngressFileBackend])],
+  providers: [IngressFileDbService],
+  exports: [IngressFileDbService],
+})
+export class IngressFileDbModule {}

backend/src/collections/ingress-file-db/ingress-file-db.service.ts

@@ -0,0 +1,14 @@
+import { Injectable, Logger } from '@nestjs/common';
+import { InjectRepository } from '@nestjs/typeorm';
+import { Repository } from 'typeorm';
+import { EIngressFileBackend } from '../../database/entities/ingress-file.entity';
+
+@Injectable()
+export class IngressFileDbService {
+  private readonly logger = new Logger(IngressFileDbService.name);
+
+  constructor(
+    @InjectRepository(EIngressFileBackend)
+    private readonly ingressFileRepo: Repository<EIngressFileBackend>,
+  ) {}
+}

backend/src/config/early/bull.config.service.ts

@@ -0,0 +1,25 @@
+import {
+  BullRootModuleOptions,
+  SharedBullConfigurationFactory
+} from '@nestjs/bull';
+import { Injectable } from '@nestjs/common';
+import { RedisConfigService } from './redis.config.service';
+
+@Injectable()
+export class BullConfigService implements SharedBullConfigurationFactory {
+  constructor(private readonly redisConfig: RedisConfigService) {}
+
+  async createSharedConfiguration(): Promise<BullRootModuleOptions> {
+    const options: BullRootModuleOptions = {
+      url: this.redisConfig.getRedisUrl(),
+      redis: {
+        lazyConnect: false,
+      },
+      defaultJobOptions: {
+        attempts: 3,
+        removeOnFail: true,
+      },
+    };
+    return options;
+  }
+}

backend/src/config/early/early-config.module.ts

@@ -1,9 +1,11 @@
 import { Module } from '@nestjs/common';
 import { ConfigModule } from '@nestjs/config';
 import { AuthConfigService } from './auth.config.service';
+import { BullConfigService } from './bull.config.service';
 import { EarlyJwtConfigService } from './early-jwt.config.service';
 import { HostConfigService } from './host.config.service';
 import { MultipartConfigService } from './multipart.config.service';
+import { RedisConfigService } from './redis.config.service';
 import { ServeStaticConfigService } from './serve-static.config.service';
 import { TypeOrmConfigService } from './type-orm.config.service';
 
@@ -21,6 +23,8 @@ import { TypeOrmConfigService } from './type-orm.config.service';
     HostConfigService,
     AuthConfigService,
     MultipartConfigService,
+    RedisConfigService,
+    BullConfigService,
   ],
   exports: [
     ConfigModule,
@@ -30,6 +34,8 @@ import { TypeOrmConfigService } from './type-orm.config.service';
     HostConfigService,
     AuthConfigService,
     MultipartConfigService,
+    RedisConfigService,
+    BullConfigService,
   ],
 })
 export class EarlyConfigModule {}

backend/src/config/early/redis.config.service.ts

@@ -0,0 +1,20 @@
+import { Injectable, Logger } from '@nestjs/common';
+import { ConfigService } from '@nestjs/config';
+import { ParseString } from 'picsur-shared/dist/util/parse-simple';
+import { EnvPrefix } from '../config.static';
+
+@Injectable()
+export class RedisConfigService {
+  private readonly logger = new Logger(RedisConfigService.name);
+
+  constructor(private readonly configService: ConfigService) {
+    this.logger.log('Redis URL: ' + this.getRedisUrl());
+  }
+
+  public getRedisUrl(): string {
+    return ParseString(
+      this.configService.get(`${EnvPrefix}REDIS_URL`),
+      'redis://localhost:6379',
+    );
+  }
+}

backend/src/config/early/type-orm.config.service.ts

@@ -6,6 +6,7 @@ import { EntityList } from '../../database/entities';
 import { MigrationList } from '../../database/migrations';
 import { DefaultName, EnvPrefix } from '../config.static';
 import { HostConfigService } from './host.config.service';
+import { RedisConfigService } from './redis.config.service';
 
 @Injectable()
 export class TypeOrmConfigService implements TypeOrmOptionsFactory {
@@ -14,6 +15,7 @@ export class TypeOrmConfigService implements TypeOrmOptionsFactory {
   constructor(
     private readonly configService: ConfigService,
     private readonly hostService: HostConfigService,
+    private readonly redisConfig: RedisConfigService,
   ) {
     const varOptions = this.getTypeOrmServerOptions();
 
@@ -66,6 +68,12 @@ export class TypeOrmConfigService implements TypeOrmOptionsFactory {
         entitiesDir: 'src/database/entities',
       },
 
+      cache: {
+        duration: 60000,
+        type: 'ioredis',
+        options: this.redisConfig.getRedisUrl(),
+      },
+
       ...varOptions,
     } as TypeOrmModuleOptions;
   }

backend/src/consumers/consumers.module.ts

@@ -0,0 +1,14 @@
+import { BullModule } from '@nestjs/bull';
+import { Module } from '@nestjs/common';
+import { IngestConsumer } from './ingest.consumer';
+
+@Module({
+  imports: [
+    BullModule.registerQueue({
+      name: 'image-ingest',
+    }),
+  ],
+  providers: [IngestConsumer],
+  exports: [BullModule],
+})
+export class ConsumersModule {}

backend/src/consumers/ingest.consumer.ts

@@ -0,0 +1,18 @@
+import { OnQueueError, Process, Processor } from '@nestjs/bull';
+import { Logger } from '@nestjs/common';
+import type { Job } from 'bull';
+
+@Processor('image-ingest')
+export class IngestConsumer {
+  private readonly logger = new Logger(IngestConsumer.name);
+
+  @Process()
+  async processJob(job: Job) {
+    console.log('processJob', job);
+  }
+
+  @OnQueueError()
+  async handleError(error: any) {
+    this.logger.error(error);
+  }
+}

backend/src/database/entities/index.ts

@@ -2,6 +2,7 @@ import { EApiKeyBackend } from './apikey.entity';
 import { EImageDerivativeBackend } from './images/image-derivative.entity';
 import { EImageFileBackend } from './images/image-file.entity';
 import { EImageBackend } from './images/image.entity';
+import { EIngressFileBackend } from './ingress-file.entity';
 import { ESysPreferenceBackend } from './system/sys-preference.entity';
 import { ESystemStateBackend } from './system/system-state.entity';
 import { EUsrPreferenceBackend } from './system/usr-preference.entity';
@@ -18,4 +19,5 @@ export const EntityList = [
   EUsrPreferenceBackend,
   EApiKeyBackend,
   ESystemStateBackend,
+  EIngressFileBackend,
 ];

backend/src/database/entities/ingress-file.entity.ts

@@ -0,0 +1,16 @@
+import { Column, Entity, PrimaryGeneratedColumn } from 'typeorm';
+
+@Entity()
+export class EIngressFileBackend {
+  @PrimaryGeneratedColumn('uuid')
+  id: string;
+
+  @Column({ nullable: false })
+  filename: string;
+
+  @Column({ type: 'bytea', nullable: false })
+  data: Buffer;
+
+  @Column({ nullable: false, default: false })
+  in_use: boolean;
+}

backend/src/routes/api/experiment/experiment.controller.ts

@@ -1,20 +1,23 @@
-import { WebSocketGateway } from '@nestjs/websockets';
+import { InjectQueue } from '@nestjs/bull';
+import { Controller, Get } from '@nestjs/common';
+import type { Queue } from 'bull';
+import { NoPermissions } from '../../../decorators/permissions.decorator';
+import { ReturnsAnything } from '../../../decorators/returns.decorator';
 
-
-@WebSocketGateway({
-  namespace: 'experiment',
-})
+@Controller('api/experiment')
+@NoPermissions()
 export class ExperimentController {
-  constructor() {
-    console.log('ExperimentController created');
-  }
+  constructor(
 
-  // @SubscribeMessage('test')
-  // async testRoute(@MessageBody() data: any): Promise<WsResponse> {
-  //   console.log('testRoute', data);
-  //   return Promise.resolve({
-  //     event: 'test',
-  //     data: Buffer.from('Hello'),
-  //   })
-  // }
+    @InjectQueue('image-ingest') private readonly ingestQueue: Queue,
+  ) {}
+
+  @Get()
+  @ReturnsAnything()
+  async testRoute(): Promise<any> {
+    this.ingestQueue.add({ foo: Buffer.from("aaaaaheleool") });
+
+
+    return 'ok';
+  }
 }

backend/src/routes/api/experiment/experiment.module.ts

@@ -1,11 +1,12 @@
 import { Module } from '@nestjs/common';
+import { ConsumersModule } from '../../../consumers/consumers.module';
 import { ExperimentController } from './experiment.controller';
 
 // This is comletely useless module, but is used for testing
 // TODO: remove when out of beta
 
 @Module({
-  imports: [],
-  providers: [ExperimentController]
+  imports: [ConsumersModule],
+  controllers: [ExperimentController]
 })
 export class ExperimentModule {}