Fair Post Distribution Algorithm

postDistribution.ts

import { PrismaClient, Prisma, DistributionStatus, UserRole } from '@prisma/client';

const prisma = new PrismaClient();

export class PostDistributionManager {
  private distributionIntervals = [
    1, // 1 day 
    3,  // 3 day
    7,  // week
    14, 
    30 
  ];

  // Create a new post and schedule initial distribution
  async createPost(data: {
    title: string;
    content: string;
    authorId: string;
    tags?: string[];
  }) {
    // Verify user exists and can post
    const user = await prisma.user.findUnique({
      where: { id: data.authorId },
      select: {
        id: true,
        stripeSubscriptionId: true,
        stripeCurrentPeriodEnd: true,
        role: true
      }
    });

    if (!user) throw new Error('User not found');

    // Check subscription status (if not admin)
    // if (user.role !== 'ADMIN') {
    //   const isSubscribed = user.stripeSubscriptionId &&
    //     user.stripeCurrentPeriodEnd &&
    //     user.stripeCurrentPeriodEnd > new Date();

    //   if (!isSubscribed) {
    //     throw new Error('Active subscription required to create posts');
    //   }
    // }

    // Create post with tags
    const post = await prisma.post.create({
      data: {
        title: data.title,
        content: data.content,
        authorId: data.authorId,
        published: true,
        tags: data.tags ? {
          connectOrCreate: data.tags.map(tag => ({
            where: { name: tag },
            create: { name: tag }
          }))
        } : undefined
      },
      include: {
        author: {
          select: {
            name: true,
            image: true
          }
        },
        tags: true
      }
    });
    console.log(post,"post");
    // Schedule initial distributions
    await this.scheduleDistributions(post.id);
    return post;
  }

  // Schedule distributions for a post
  private async scheduleDistributions(postId: string, batchSize = 100) {
    // Get users who haven't received this post yet
    const users = await prisma.user.findMany({
      where: {
        NOT: {
          OR: [
            {
              receivedPosts: {
                some: { postId }
              }
            },
            {
              authoredPosts: {
                some: { id: postId }
              }
            }
          ]
        },
        // Only distribute to users with verified email
        // emailVerified: { not: null }
      },
      orderBy: {
        activityScore: 'desc'
      },
      take: batchSize,
      select: {
        id: true,
        activityScore: true
      }
    });


   console.log(users,"users");

    if (users.length === 0) return;

    // Create distribution records
    const distributions = users.map(user => {
      const scheduledFor = this.calculateDeliveryTime(user.activityScore);
      return {
        postId,
        userId: user.id,
        status: 'PENDING' as DistributionStatus,
        scheduledFor
      };
    });


    await prisma.distribution.createMany({
      data: distributions,
      skipDuplicates: true
    });
  }

  // Calculate delivery time based on activity score
  private calculateDeliveryTime(activityScore: number): Date {
    const now = new Date();
    const baseDelay = this.distributionIntervals[0];
    const activityMultiplier = (10 - activityScore) / 10;
    const delayDays = Math.max(baseDelay, baseDelay + (baseDelay * activityMultiplier));
    
    now.setDate(now.getDate() + delayDays);
    return now;
  }

  // Process pending distributions (run this with a cron job)
  async processPendingDistributions() {
    const now = new Date();
    
    const pendingDistributions = await prisma.distribution.findMany({
      where: {
        status: 'PENDING',
        scheduledFor: {
          lte: now
        }
      },
      select: {
        id: true,
        postId: true,
        userId: true
      },
      take: 1000 // Process in batches
    });

    if (pendingDistributions.length === 0) return;

    // Update distributions to delivered
    await prisma.$transaction(async (tx) => {
      // Mark distributions as delivered
      await tx.distribution.updateMany({
        where: {
          id: {
            in: pendingDistributions.map(d => d.id)
          }
        },
        data: {
          status: 'DELIVERED',
          deliveredAt: now
        }
      });
      
      // Schedule next batch of distributions
      await Promise.all(pendingDistributions.map(d => 
        this.scheduleDistributions(d.postId)
      ));
    });

    return pendingDistributions.length;
  }
  // Get user's feed
  async getUserFeed(userId: string, options: { page?: number, pageSize?: number, tag?: string } = {}) {
    const { page = 1, pageSize = 20, tag } = options;
    const skip = (page - 1) * pageSize;

    return await prisma.distribution.findMany({
      where: {
        userId,
        status: 'DELIVERED',
        post: tag ? {
          tags: {
            some: { name: tag }
          }
        } : undefined
      },
      include: {
        post: {
          include: {
            author: {
              select: {
                id: true,
                name: true,
                image: true
              }
            },
            tags: true,
            _count: {
              select: {
                interactions: {
                  where: {
                    type: 'like'
                  }
                }
              }
            }
          }
        }
      },
      orderBy: {
        deliveredAt: 'desc'
      },
      skip,
      take: pageSize
    });
  }
  
  // Record user interaction and update activity score
  async recordInteraction(data: { userId: string, postId: string, type: 'view' | 'like' | 'comment' | 'share' }) {
    const { userId, postId, type } = data;

    // Score weights for different interaction types
    const scoreChanges = {
      view: 0.1,
      like: 0.2,
      comment: 0.5,
      share: 1.0
    };

    return await prisma.$transaction(async (tx) => {
      // Record the interaction
      await tx.postInteraction.create({
        data: {
          type,
          userId,
          postId
        }
      });

      // If this is a view of a delivered post, mark it as seen
      if (type === 'view') {
        await tx.distribution.updateMany({
          where: {
            postId,
            userId,
            status: 'DELIVERED'
          },
          data: {
            status: 'SEEN'
          }
        });
      }

      // Update user activity score
      const user = await tx.user.findUnique({
        where: { id: userId }
      });

      if (user) {
        const scoreChange = scoreChanges[type] || 0;
        const newScore = Math.min(10, Math.max(1, user.activityScore + scoreChange));

        await tx.user.update({
          where: { id: userId },
          data: {
            activityScore: newScore,
            lastActive: new Date()
          }
        });
      }
    });
  }

  async getPostAnalytics(postId: string) {
    return await prisma.post.findUnique({
      where: { id: postId },
      include: {
        _count: {
          select: {
            distributions: {
              where: {
                OR: [
                  { status: 'DELIVERED' },
                  { status: 'SEEN' },
                  { status: 'INTERACTED' }
                ]
              }
            },
            interactions: {
              where: {
                type: 'like'
              }
            }
          }
        },
        interactions: {
          select: {
            type: true,
            createdAt: true
          }
        },
        distributions: {
          select: {
            status: true
          }
        }
      }
    });
  }
}