feat: replace Google Pub/Sub with Redis (#67)

.env.production

@@ -15,3 +15,6 @@ JWT_ACCESS_TOKEN_SECRET=
 JWT_ACCESS_TOKEN_EXPIRATION_TIME=86400
 JWT_REFRESH_TOKEN_SECRET=
 JWT_REFRESH_TOKEN_EXPIRATION_TIME=108000
+
+REDIS_HOST=localhost
+REDIS_PORT=6379

charts/redis.yaml

@@ -0,0 +1,9 @@
+image:
+  repository: redis
+  tag: 6.2-alpine
+  pullPolicy: Always
+
+service:
+  type: LoadBalancer
+  targetPort: 6379
+  port: 6379

package.json

@@ -38,7 +38,6 @@
     }
   },
   "dependencies": {
-    "@axelspringer/graphql-google-pubsub": "^2.1.0",
     "@nestjs/apollo": "^10.0.8",
     "@nestjs/common": "8.4.4",
     "@nestjs/config": "^2.0.0",
@@ -61,6 +60,7 @@
     "googleapis": "^100.0.0",
     "graphql": "^16.3.0",
     "graphql-fields-list": "^2.2.4",
+    "graphql-redis-subscriptions": "^2.4.2",
     "graphql-tools": "8.2.5",
     "graphql-ws": "^5.8.2",
     "luxon": "^2.3.2",

src/app.module.ts

@@ -13,62 +13,53 @@ import { TmdbModule } from './modules/tmdb';
 import { ShowModule } from './modules/show';
 import { GoogleModule } from './modules/google';
 import { HealthModule } from './modules/health';
-import { AuthModule, AuthService } from './modules/auth';
+import { AuthModule } from './modules/auth';
 import { PreferenceModule } from './modules/preference';
 import { WatchlistModule } from './modules/watchlist';
 import { ReviewModule } from './modules/review';
 import { StatsModule } from './modules/stats';
 import { SearchModule } from './modules/search';
 import { NotificationModule } from './modules/notification';
 import { CORS_ORIGINS } from './app.constants';
-import { getCookie } from './util/cookie';
 
 @Module({
   imports: [
     ConfigModule.forRoot({
       envFilePath: ['.env', '.env.production'],
     }),
     ConfigModule.forFeature(appConfig),
-    GraphQLModule.forRootAsync<ApolloDriverConfig>({
+    GraphQLModule.forRoot<ApolloDriverConfig>({
       driver: ApolloDriver,
-      imports: [AuthModule],
-      inject: [AuthService],
-      useFactory: async (authService: AuthService) => ({
-        autoSchemaFile: join(process.cwd(), 'src/schema.gql'),
-        playground: false,
-        // installSubscriptionHandlers: true,
-        async context({ extra }) {
-          let user = null;
+      autoSchemaFile: join(process.cwd(), 'src/schema.gql'),
+      playground: false,
+      context: ({ req, extra, connectionParams }) => {
+        if (extra) {
+          return {
+            req: {
+              ...extra.request,
+              headers: {
+                ...extra.request.headers,
+                authorization: connectionParams.authorization,
+              },
+            },
+          };
+        }
 
-          if (extra) {
-            try {
-              const authToken = getCookie(
-                extra.request.headers.cookie,
-                'Authentication',
-              );
-
-              user = await authService.getUserFromJwtToken(authToken);
-            } catch (e) {
-              console.log('e', e);
-            }
-          }
-
-          return { user };
-        },
-        subscriptions: {
-          'graphql-ws': {
-            path: '/subscriptions',
-          },
-        },
-        plugins: [
-          ApolloServerPluginLandingPageLocalDefault(),
-          ApolloServerPluginInlineTrace(),
-        ],
-        cors: {
-          credentials: 'include',
-          origin: CORS_ORIGINS,
+        return { req };
+      },
+      subscriptions: {
+        'graphql-ws': {
+          path: '/subscriptions',
         },
-      }),
+      },
+      plugins: [
+        ApolloServerPluginLandingPageLocalDefault(),
+        ApolloServerPluginInlineTrace(),
+      ],
+      cors: {
+        credentials: 'include',
+        origin: CORS_ORIGINS,
+      },
     }),
     HealthModule,
     TmdbModule,

src/modules/auth/auth.module.ts

@@ -10,7 +10,6 @@ import {
   JwtRefreshTokenStrategy,
 } from './strategies';
 import { authConfig } from './auth.config';
-import { AuthController } from './controllers';
 import { AuthResolver } from './resolvers';
 
 @Module({
@@ -28,6 +27,5 @@ import { AuthResolver } from './resolvers';
     JwtStrategy,
     JwtRefreshTokenStrategy,
   ],
-  controllers: [AuthController],
 })
 export class AuthModule {}

src/modules/auth/entities/index.ts

@@ -0,0 +1 @@
+export { Token } from './token';

src/modules/auth/entities/token.ts

@@ -0,0 +1,11 @@
+import 'reflect-metadata';
+import { Field, ObjectType } from '@nestjs/graphql';
+
+@ObjectType()
+export class Token {
+  @Field()
+  accessToken: string;
+
+  @Field()
+  refreshToken: string;
+}

src/modules/auth/resolvers/auth.resolvers.ts

@@ -27,11 +27,11 @@ export class AuthResolver {
   @Mutation(() => User)
   @UseGuards(GraphqlJwtRefreshGuard)
   async refresh(@Context() { req }: { req: RequestWithUser }) {
-    const accessTokenCookie = this.authService.getCookieWithJwtAccessToken(
+    const accessToken = this.authService.getCookieWithJwtAccessToken(
       req.user.id,
     );
 
-    req.res.setHeader('Set-Cookie', accessTokenCookie);
+    req.res.setHeader('Set-Cookie', accessToken.cookie);
 
     return req.user;
   }

src/modules/auth/services/auth.service.ts

@@ -37,11 +37,15 @@ export class AuthService {
       secret: this.config.jwtAccessTokenSecret,
       expiresIn: `${this.config.jwtAccessTokenExirationTime}s`,
     });
-
-    return createCookie(AUTHENTICATION_COOKIE, {
+    const cookie = createCookie(AUTHENTICATION_COOKIE, {
       value: token,
       maxAge: this.config.jwtAccessTokenExirationTime,
     });
+
+    return {
+      cookie,
+      token,
+    };
   }
 
   public getCookieWithJwtRefreshToken(userId: number) {

src/modules/auth/strategies/jwt.strategy.ts

@@ -5,7 +5,6 @@ import { ConfigService, ConfigType } from '@nestjs/config';
 import { Request } from 'express';
 import { UserService } from '../../user/services';
 import { authConfig } from '../auth.config';
-import { AUTHENTICATION_COOKIE } from '../auth.constants';
 
 interface TokenPayload {
   userId: number;
@@ -22,7 +21,9 @@ export class JwtStrategy extends PassportStrategy(Strategy) {
   ) {
     super({
       jwtFromRequest: ExtractJwt.fromExtractors([
-        (request: Request) => request?.cookies?.[AUTHENTICATION_COOKIE],
+        (request: Request) => {
+          return request.headers.authorization;
+        },
       ]),
       secretOrKey: config.jwtAccessTokenSecret,
     });

src/modules/google/resolvers/google.resolvers.ts

@@ -4,23 +4,25 @@ import { Request } from 'express';
 import { Injectable } from '@nestjs/common';
 import { User } from '../../user/entities';
 import { GoogleService } from '../services';
+import { Void } from '../../../util/void';
 import { JoinWithGoogleInput } from './input';
 
 @Injectable()
 @Resolver(User)
 export class GoogleResolver {
   constructor(private readonly googleService: GoogleService) {}
 
-  @Mutation(() => User)
+  @Mutation(() => Void)
   async joinWithGoogle(
     @Args('input') input: JoinWithGoogleInput,
     @Context() { req }: { req: Request },
-  ) {
-    const { user, refreshTokenCookie, accessTokenCookie } =
-      await this.googleService.authenticate(input.token);
+  ): Promise<Void> {
+    const { accessToken, refreshToken } = await this.googleService.authenticate(
+      input.token,
+    );
 
-    req.res.setHeader('Set-Cookie', [accessTokenCookie, refreshTokenCookie]);
+    req.res.setHeader('Set-Cookie', [accessToken.cookie, refreshToken.cookie]);
 
-    return user;
+    return {};
   }
 }

src/modules/google/services/google.service.ts

@@ -60,28 +60,23 @@ export class GoogleService {
   }
 
   private async getCookiesForUser(user: User) {
-    const accessTokenCookie = this.authService.getCookieWithJwtAccessToken(
-      user.id,
-    );
-    const { cookie: refreshTokenCookie, token: refreshToken } =
-      this.authService.getCookieWithJwtRefreshToken(user.id);
+    const accessToken = this.authService.getCookieWithJwtAccessToken(user.id);
+    const refreshToken = this.authService.getCookieWithJwtRefreshToken(user.id);
 
-    await this.userService.setCurrentRefreshToken(refreshToken, user.id);
+    await this.userService.setCurrentRefreshToken(refreshToken.token, user.id);
 
     return {
-      accessTokenCookie,
-      refreshTokenCookie,
+      accessToken,
+      refreshToken,
     };
   }
 
   private async handleRegisteredUser(user: User) {
-    const { accessTokenCookie, refreshTokenCookie } =
-      await this.getCookiesForUser(user);
+    const { accessToken, refreshToken } = await this.getCookiesForUser(user);
 
     return {
-      accessTokenCookie,
-      refreshTokenCookie,
-      user,
+      accessToken,
+      refreshToken,
     };
   }
 

src/modules/notification/notification.config.ts

@@ -0,0 +1,7 @@
+import { registerAs } from '@nestjs/config';
+import { NOTIFICATION_MODULE_ID } from './notification.constants';
+
+export const notificationConfig = registerAs(NOTIFICATION_MODULE_ID, () => ({
+  redisHost: process.env.REDIS_HOST,
+  redisPort: process.env.REDIS_PORT,
+}));

src/modules/notification/notification.constants.ts

@@ -0,0 +1 @@
+export const NOTIFICATION_MODULE_ID = 'notification';

src/modules/notification/notification.module.ts

@@ -5,14 +5,14 @@ import { PrismaModule } from '../prisma';
 import { WatchlistModule } from '../watchlist';
 import { TmdbModule } from '../tmdb';
 import { GoogleModule } from '../google';
-import { googleConfig } from '../google/google.config';
 import * as resolvers from './resolvers';
 import * as services from './services';
 import { NotificationSchedulerService } from './services';
+import { notificationConfig } from './notification.config';
 
 @Module({
   imports: [
-    ConfigModule.forFeature(googleConfig),
+    ConfigModule.forFeature(notificationConfig),
     ScheduleModule.forRoot(),
     PrismaModule,
     WatchlistModule,

src/modules/notification/resolvers/notification.resolvers.ts

@@ -60,17 +60,14 @@ export class NotificationResolver {
   }
 
   @Subscription(() => [Notification], {
-    filter({ data }, variables, { user }) {
-      const { userId } = JSON.parse(data);
-
+    filter({ userId }, variables, { req: { user } }) {
       return userId === user.id;
     },
-    resolve(this: NotificationResolver, { data }) {
-      const { notificationIds } = JSON.parse(data);
-
+    resolve(this: NotificationResolver, { notificationIds }) {
       return this.notificationService.listNotificationsByIds(notificationIds);
     },
   })
+  @UseGuards(GraphqlJwtAuthGuard)
   notificationsAdded() {
     return this.notificationPubsubService.getAsyncIterator();
   }