Subscriptions

/**
 * Injectable service for managing GraphQL subscriptions
 * Handles WebSocket connections, subscription lifecycle, and pub/sub messaging
 */
export class GqlSubscriptionService {
  /**
   * Start subscription server with WebSocket support
   * @param options - Subscription server configuration
   * @returns Promise that resolves when server is started
   */
  start(options: SubscriptionServerOptions): Promise<void>;
  
  /**
   * Stop subscription server and close all connections
   * @returns Promise that resolves when server is stopped
   */
  stop(): Promise<void>;
  
  /**
   * Get active subscription count
   * @returns Number of active subscriptions
   */
  getActiveSubscriptionCount(): number;
  
  /**
   * Publish message to subscription channel
   * @param channel - Channel name to publish to
   * @param payload - Data to publish
   * @returns Promise that resolves when message is published
   */
  publish(channel: string, payload: any): Promise<void>;
  
  /**
   * Subscribe to channel for updates
   * @param channel - Channel name to subscribe to
   * @returns Async iterator for subscription events
   */
  subscribe(channel: string): AsyncIterator<any>;
  
  /**
   * Unsubscribe from channel
   * @param channel - Channel name to unsubscribe from
   * @param subscriptionId - Unique subscription identifier
   * @returns Promise that resolves when unsubscribed
   */
  unsubscribe(channel: string, subscriptionId: string): Promise<void>;
}

import { Injectable } from "@nestjs/common";
import { GqlSubscriptionService } from "@nestjs/graphql";
import { PubSub } from "graphql-subscriptions";

@Injectable()
export class NotificationService {
  private pubSub = new PubSub();
  
  constructor(private subscriptionService: GqlSubscriptionService) {}
  
  // Publish notification
  async publishNotification(userId: string, message: string): Promise<void> {
    await this.subscriptionService.publish(`user_${userId}`, {
      notification: { message, timestamp: new Date() }
    });
  }
  
  // Subscribe to user notifications
  subscribeToUserNotifications(userId: string): AsyncIterator<any> {
    return this.subscriptionService.subscribe(`user_${userId}`);
  }
}

// Subscription resolver
@Resolver()
export class SubscriptionResolver {
  constructor(private notificationService: NotificationService) {}
  
  @Subscription(() => Notification)
  notifications(@Args('userId') userId: string): AsyncIterator<Notification> {
    return this.notificationService.subscribeToUserNotifications(userId);
  }
  
  @Subscription(() => String, {
    filter: (payload, variables) => {
      return payload.userId === variables.userId;
    },
  })
  messageAdded(@Args('userId') userId: string): AsyncIterator<string> {
    return this.pubSub.asyncIterator('messageAdded');
  }
}

/**
 * General subscription configuration
 * Union type supporting multiple WebSocket transport libraries
 */
type SubscriptionConfig = 
  | GraphQLWsSubscriptionsConfig 
  | GraphQLSubscriptionTransportWsConfig 
  | boolean;

/**
 * Configuration for graphql-ws WebSocket subscriptions (recommended)
 * Modern WebSocket transport with improved performance and features
 */
interface GraphQLWsSubscriptionsConfig {
  /** WebSocket server host */
  host?: string;
  /** WebSocket server port */
  port?: number;
  /** WebSocket endpoint path */
  path?: string;
  /** Connection initialization options */
  connectionInitWaitTimeout?: number;
  /** Keep-alive interval in milliseconds */
  keepAlive?: number;
  /** Custom context function for subscriptions */
  context?: (ctx: any) => any;
  /** Connection callback functions */
  onConnect?: (connectionParams: any) => any;
  onDisconnect?: (websocket: any, context: any) => any;
  /** Custom schema for subscriptions */
  schema?: GraphQLSchema;
  /** Enable introspection for subscriptions */
  introspection?: boolean;
}

/**
 * Configuration for subscriptions-transport-ws (legacy)
 * Legacy WebSocket transport, use graphql-ws for new projects
 */
interface GraphQLSubscriptionTransportWsConfig {
  /** WebSocket server host */
  host?: string;
  /** WebSocket server port */  
  port?: number;
  /** WebSocket endpoint path */
  path?: string;
  /** Connection timeout in milliseconds */
  timeout?: number;
  /** Keep-alive interval */
  keepAlive?: number;
  /** Connection callback functions */
  onConnect?: (connectionParams: any, websocket: any, context: any) => any;
  onDisconnect?: (websocket: any, context: any) => any;
  onOperation?: (message: any, params: any, websocket: any) => any;
  onOperationComplete?: (websocket: any, opId: string) => any;
}

/**
 * Subscription server configuration options
 */
interface SubscriptionServerOptions {
  /** HTTP server instance to attach WebSocket server to */
  server: any;
  /** GraphQL schema for subscriptions */
  schema: GraphQLSchema;
  /** Subscription configuration */
  subscriptions: SubscriptionConfig;
  /** GraphQL context function */
  context?: (ctx: any) => any;
}

import { Module } from "@nestjs/common";
import { GraphQLModule } from "@nestjs/graphql";
import { ApolloDriver, ApolloDriverConfig } from "@nestjs/apollo";

// Modern graphql-ws configuration
@Module({
  imports: [
    GraphQLModule.forRoot<ApolloDriverConfig>({
      driver: ApolloDriver,
      autoSchemaFile: true,
      installSubscriptionHandlers: true,
      subscriptions: {
        'graphql-ws': {
          path: '/graphql',
          onConnect: (connectionParams) => {
            console.log('Client connected:', connectionParams);
            return { user: connectionParams.user };
          },
          onDisconnect: () => {
            console.log('Client disconnected');
          },
          context: ({ connectionParams }) => ({
            user: connectionParams?.user,
          }),
        },
      },
    }),
  ],
})
export class SubscriptionModule {}

// Legacy subscriptions-transport-ws configuration
@Module({
  imports: [
    GraphQLModule.forRoot<ApolloDriverConfig>({
      driver: ApolloDriver,
      autoSchemaFile: true,
      installSubscriptionHandlers: true,
      subscriptions: {
        'subscriptions-transport-ws': {
          path: '/graphql',
          onConnect: (connectionParams, websocket, context) => {
            return { user: connectionParams.user };
          },
          keepAlive: 30000,
        },
      },
    }),
  ],
})
export class LegacySubscriptionModule {}

/**
 * Marks a resolver method as a GraphQL Subscription
 * @param name - Optional subscription name (defaults to method name)
 * @param options - Configuration options for the subscription
 */
function Subscription(name?: string, options?: SubscriptionOptions): MethodDecorator;

/**
 * Options for subscription resolvers
 */
interface SubscriptionOptions {
  /** Custom name for the subscription */
  name?: string;
  /** Description for the subscription */
  description?: string;
  /** Deprecation reason */
  deprecationReason?: string;
  /** Whether return type can be null */
  nullable?: boolean;
  /** Filter function to determine which updates to send */
  filter?: (payload: any, variables: any, context: any) => boolean | Promise<boolean>;
  /** Resolver function to transform subscription payload */
  resolve?: (payload: any, args: any, context: any, info: any) => any;
}

import { Resolver, Subscription, Args, Context } from "@nestjs/graphql";
import { PubSub } from "graphql-subscriptions";

@Resolver()
export class ChatSubscriptionResolver {
  private pubSub = new PubSub();
  
  constructor(private chatService: ChatService) {}
  
  // Basic subscription
  @Subscription(() => Message)
  messageAdded(): AsyncIterator<Message> {
    return this.pubSub.asyncIterator('messageAdded');
  }
  
  // Subscription with filtering
  @Subscription(() => Message, {
    filter: (payload, variables, context) => {
      // Only send messages from the specified room
      return payload.messageAdded.roomId === variables.roomId;
    },
  })
  messageAddedToRoom(@Args('roomId') roomId: string): AsyncIterator<Message> {
    return this.pubSub.asyncIterator('messageAdded');
  }
  
  // Subscription with payload transformation
  @Subscription(() => Notification, {
    resolve: (payload, args, context) => {
      // Transform the payload before sending to client
      return {
        id: payload.id,
        message: payload.message,
        user: context.user,
        timestamp: new Date(),
      };
    },
  })
  userNotifications(@Context() context: any): AsyncIterator<Notification> {
    const userId = context.user.id;
    return this.pubSub.asyncIterator(`userNotifications_${userId}`);
  }
  
  // Subscription with authentication
  @Subscription(() => PrivateMessage, {
    filter: (payload, variables, context) => {
      // Ensure user is authenticated and authorized
      if (!context.user) return false;
      return payload.recipientId === context.user.id;
    },
  })
  privateMessageReceived(@Context() context: any): AsyncIterator<PrivateMessage> {
    return this.pubSub.asyncIterator('privateMessage');
  }
  
  // Trigger subscription from mutation
  @Mutation(() => Message)
  async sendMessage(@Args('input') input: SendMessageInput): Promise<Message> {
    const message = await this.chatService.createMessage(input);
    
    // Trigger subscription
    await this.pubSub.publish('messageAdded', { messageAdded: message });
    
    return message;
  }
}

/**
 * Interface for pub/sub system integration
 */
interface PubSubEngine {
  /** Publish message to channel */
  publish(channel: string, payload: any): Promise<void>;
  /** Subscribe to channel */
  subscribe(channel: string, onMessage: (message: any) => void): Promise<number>;
  /** Unsubscribe from channel */
  unsubscribe(subId: number): void;
  /** Get async iterator for channel */
  asyncIterator<T>(triggers: string | string[]): AsyncIterator<T>;
}

/**
 * Redis pub/sub configuration
 */
interface RedisPubSubOptions {
  /** Redis connection options */
  connection?: {
    host: string;
    port: number;
    password?: string;
    db?: number;
  };
  /** Custom Redis client */
  publisher?: any;
  subscriber?: any;
  /** Message serialization options */
  serializer?: {
    serialize: (value: any) => string;
    deserialize: (value: string) => any;
  };
}

import { Module } from "@nestjs/common";
import { RedisPubSub } from "graphql-redis-subscriptions";
import { PubSub } from "graphql-subscriptions";
import Redis from "ioredis";

// In-memory pub/sub (development only)
@Module({
  providers: [
    {
      provide: 'PUB_SUB',
      useValue: new PubSub(),
    },
  ],
  exports: ['PUB_SUB'],
})
export class PubSubModule {}

// Redis pub/sub (production)
@Module({
  providers: [
    {
      provide: 'PUB_SUB',
      useFactory: () => {
        return new RedisPubSub({
          publisher: new Redis({
            host: 'localhost',
            port: 6379,
          }),
          subscriber: new Redis({
            host: 'localhost',
            port: 6379,
          }),
        });
      },
    },
  ],
  exports: ['PUB_SUB'],
})
export class RedisPubSubModule {}

// Using pub/sub in resolver
@Resolver()
export class NotificationResolver {
  constructor(@Inject('PUB_SUB') private pubSub: PubSubEngine) {}
  
  @Subscription(() => Notification)
  notifications(): AsyncIterator<Notification> {
    return this.pubSub.asyncIterator('notifications');
  }
  
  @Mutation(() => Boolean)
  async sendNotification(@Args('message') message: string): Promise<boolean> {
    await this.pubSub.publish('notifications', {
      notifications: { message, timestamp: new Date() }
    });
    return true;
  }
}

/**
 * Subscription authentication context
 */
interface SubscriptionContext {
  /** WebSocket connection */
  connection: any;
  /** Connection parameters from client */
  connectionParams: any;
  /** Authenticated user */
  user?: any;
  /** Additional context data */
  [key: string]: any;
}

/**
 * Subscription guard interface
 */
interface SubscriptionGuard {
  canActivate(context: SubscriptionContext): boolean | Promise<boolean>;
}

import { CanActivate, Injectable, ExecutionContext } from "@nestjs/common";
import { GqlExecutionContext } from "@nestjs/graphql";

// Subscription authentication guard
@Injectable()
export class SubscriptionAuthGuard implements CanActivate {
  canActivate(context: ExecutionContext): boolean {
    const gqlContext = GqlExecutionContext.create(context);
    const ctx = gqlContext.getContext();
    
    // Check if user is authenticated
    return !!ctx.user;
  }
}

// Subscription with authentication
@Resolver()
@UseGuards(SubscriptionAuthGuard)
export class SecureSubscriptionResolver {
  @Subscription(() => PrivateMessage)
  privateMessages(@Context() context: SubscriptionContext): AsyncIterator<PrivateMessage> {
    const userId = context.user.id;
    return this.pubSub.asyncIterator(`privateMessages_${userId}`);
  }
}

// Module configuration with authentication
@Module({
  imports: [
    GraphQLModule.forRoot<ApolloDriverConfig>({
      driver: ApolloDriver,
      autoSchemaFile: true,
      subscriptions: {
        'graphql-ws': {
          onConnect: async (connectionParams) => {
            // Authenticate connection
            const token = connectionParams.authorization;
            if (!token) throw new Error('Missing auth token');
            
            const user = await this.authService.validateToken(token);
            if (!user) throw new Error('Invalid auth token');
            
            return { user };
          },
          context: ({ connectionParams }) => ({
            user: connectionParams.user,
          }),
        },
      },
    }),
  ],
})
export class AuthenticatedSubscriptionModule {}

/**
 * Subscription error handling patterns
 */
interface SubscriptionErrorHandler {
  /** Handle subscription errors */
  handleError(error: Error, context: SubscriptionContext): any;
  /** Handle connection errors */
  handleConnectionError(error: Error, connectionParams: any): any;
}

@Resolver()
export class RobustSubscriptionResolver {
  @Subscription(() => Message, {
    filter: async (payload, variables, context) => {
      try {
        // Potentially failing filter logic
        const hasPermission = await this.checkPermission(context.user, variables.roomId);
        return hasPermission;
      } catch (error) {
        console.error('Subscription filter error:', error);
        return false; // Safely exclude on error
      }
    },
    resolve: (payload, args, context) => {
      try {
        // Transform payload with error handling
        return this.transformMessage(payload.messageAdded);
      } catch (error) {
        console.error('Subscription resolve error:', error);
        return null; // Return null on transformation error
      }
    },
  })
  roomMessages(@Args('roomId') roomId: string): AsyncIterator<Message> {
    return this.pubSub.asyncIterator(`room_${roomId}`);
  }
}