RankFlow AI — GBP & Social Media Pipeline Documentation

Version: 1.0.0
GBP Path: src/server/services/gbp/
Social Path: src/server/services/social/
Auth: Google OAuth (GBP) + Composio (Social)

---

1. GBP OAuth Flow#

Authorization URL Generation#

// src/server/services/gbp/auth.ts

import { google } from "googleapis";
import { encrypt } from "@/lib/crypto";

const oauth2Client = new google.auth.OAuth2(
  process.env.GOOGLE_CLIENT_ID,
  process.env.GOOGLE_CLIENT_SECRET,
  `${process.env.APP_URL}/api/webhooks/google/oauth-callback`
);

export async function getAuthUrl(practiceId: string): Promise<string> {
  const state = Buffer.from(JSON.stringify({ practiceId })).toString("base64");
  
  return oauth2Client.generateAuthUrl({
    access_type: "offline",
    scope: [
      "https://www.googleapis.com/auth/business.manage",
      "https://www.googleapis.com/auth/userinfo.email",
      "https://www.googleapis.com/auth/userinfo.profile",
    ],
    prompt: "consent",
    state,
    include_granted_scopes: true,
  });
}

OAuth Callback Handler#

// src/app/api/webhooks/google/oauth-callback/route.ts

export async function GET(request: Request) {
  const { searchParams } = new URL(request.url);
  const code = searchParams.get("code");
  const state = searchParams.get("state");
  
  if (!code || !state) {
    return new Response("Invalid callback", { status: 400 });
  }
  
  const { practiceId } = JSON.parse(Buffer.from(state, "base64").toString());
  
  // Exchange code for tokens
  const { tokens } = await oauth2Client.getToken(code);
  
  // Get user info
  oauth2Client.setCredentials(tokens);
  const oauth2 = google.oauth2({ version: "v2", auth: oauth2Client });
  const { data: userInfo } = await oauth2.userinfo.get();
  
  // Store encrypted tokens
  await db.gbpAccount.create({
    data: {
      practiceId,
      accountEmail: userInfo.email!,
      accessToken: encrypt(tokens.access_token!),
      refreshToken: encrypt(tokens.refresh_token!),
      tokenExpiresAt: new Date(Date.now() + (tokens.expiry_date || 3600 * 1000)),
      scope: (tokens.scope || "").split(" ").filter(Boolean),
    },
  });
  
  // Trigger sync workflow
  await inngest.send({
    name: "gbp/sync-locations",
    data: { practice_id: practiceId },
  });
  
  // Redirect to dashboard
  return NextResponse.redirect(`${process.env.APP_URL}/dashboard/gbp?connected=true`);
}

---

2. GBP Token Management#

Automatic Refresh#

// src/server/services/gbp/auth.ts

export async function getClient(accountId: string) {
  const account = await db.gbpAccount.findUnique({ where: { id: accountId } });
  if (!account) throw new Error("GBP account not found");
  
  const accessToken = decrypt(account.accessToken);
  const refreshToken = decrypt(account.refreshToken);
  
  // Refresh if expiring within 5 minutes
  if (account.tokenExpiresAt.getTime() - Date.now() < 5 * 60 * 1000) {
    oauth2Client.setCredentials({ refresh_token: refreshToken });
    const { tokens } = await oauth2Client.refreshAccessToken();
    
    await db.gbpAccount.update({
      where: { id: accountId },
      data: {
        accessToken: encrypt(tokens.access_token!),
        ...(tokens.refresh_token ? { refreshToken: encrypt(tokens.refresh_token) } : {}),
        tokenExpiresAt: new Date(Date.now() + (tokens.expiry_date || 3600 * 1000)),
      },
    });
    
    oauth2Client.setCredentials({ access_token: tokens.access_token });
  } else {
    oauth2Client.setCredentials({ access_token: accessToken });
  }
  
  return google.mybusinessbusinessinformation({ version: "v1", auth: oauth2Client });
}

Token Refresh Cron Job#

// src/server/bullmq/processors/token-refresh.ts

export async function tokenRefreshProcessor(job: Job) {
  const accounts = await db.gbpAccount.findMany({
    where: {
      isActive: true,
      tokenExpiresAt: { lt: new Date(Date.now() + 24 * 60 * 60 * 1000) },
    },
  });
  
  const results = await Promise.allSettled(
    accounts.map(async (account) => {
      try {
        await getClient(account.id); // This triggers refresh
        return { accountId: account.id, success: true };
      } catch (error) {
        // Notify if refresh fails
        await email.send({
          to: "admin@rankflow.ai",
          subject: `GBP Token Refresh Failed: ${account.accountEmail}`,
          body: `Practice: ${account.practiceId}\nError: ${(error as Error).message}`,
        });
        return { accountId: account.id, success: false, error: (error as Error).message };
      }
    })
  );
  
  return { refreshed: results.filter(r => r.status === "fulfilled").length };
}

---

3. GBP Operations#

Post Creation#

// src/server/services/gbp/client.ts

export async function createPost(
  accountId: string,
  locationId: string,
  content: string,
  options: {
    mediaUrls?: string[];
    ctaType?: "BOOK" | "CALL" | "LEARN_MORE" | "SIGN_UP" | "ORDER";
    topicType?: "STANDARD" | "OFFER" | "EVENT";
    scheduledFor?: Date;
  }
): Promise<{ postId: string; status: string }> {
  const client = await getClient(accountId);
  
  const postBody: any = {
    languageCode: "en-IN",
    topicType: options.topicType || "STANDARD",
    body: content,
  };
  
  if (options.ctaType) {
    postBody.callToAction = {
      actionType: options.ctaType,
      url: options.ctaType === "BOOK" ? "https://booking.url" : undefined,
    };
  }
  
  if (options.mediaUrls?.length) {
    postBody.media = options.mediaUrls.map(url => ({
      mediaFormat: url.endsWith(".mp4") ? "VIDEO" : "PHOTO",
      sourceUrl: url,
    }));
  }
  
  const response = await client.accounts.locations.localPosts.create({
    parent: `accounts/${accountId}/locations/${locationId}`,
    requestBody: postBody,
  });
  
  return {
    postId: response.data.name!,
    status: options.scheduledFor ? "SCHEDULED" : "PUBLISHED",
  };
}

Review Reply#

export async function replyToReview(
  accountId: string,
  locationId: string,
  reviewId: string,
  replyText: string
): Promise<void> {
  const client = await getClient(accountId);
  
  await client.accounts.locations.reviews.updateReply({
    name: `accounts/${accountId}/locations/${locationId}/reviews/${reviewId}`,
    requestBody: {
      comment: replyText,
      updateTime: new Date().toISOString(),
    },
  });
}

Insights Fetch#

export async function getInsights(
  accountId: string,
  locationId: string,
  dateFrom: Date,
  dateTo: Date
): Promise<GbpInsightData> {
  const client = await getClient(accountId);
  
  const response = await client.accounts.locations.getDailyMetricsTimeSeries({
    name: `accounts/${accountId}/locations/${locationId}`,
    dailyMetricsTimeSeries: {
      startDate: { year: dateFrom.getFullYear(), month: dateFrom.getMonth() + 1, day: dateFrom.getDate() },
      endDate: { year: dateTo.getFullYear(), month: dateTo.getMonth() + 1, day: dateTo.getDate() },
    },
  });
  
  return parseInsightsResponse(response.data);
}

---

4. GBP Rate Limiting#

Quota Limits#

Implementation#

// src/server/services/gbp/ratelimit.ts

import { redis } from "@/lib/redis";

const QUOTA_LIMITS: Record<string, { daily: number; perMinute: number }> = {
  "business-information": { daily: 10000, perMinute: 100 },
  posts: { daily: 500, perMinute: 30 },
  reviews: { daily: 1000, perMinute: 60 },
  photos: { daily: 1000, perMinute: 60 },
  insights: { daily: 500, perMinute: 30 },
  qa: { daily: 500, perMinute: 30 },
};

export async function checkGbpQuota(
  operation: keyof typeof QUOTA_LIMITS,
  accountId: string
): Promise<{ allowed: boolean; remaining: number }> {
  const limit = QUOTA_LIMITS[operation];
  const today = new Date().toISOString().slice(0, 10);
  const dailyKey = `gbp:quota:${operation}:${accountId}:${today}`;
  const minuteKey = `gbp:ratelimit:${operation}:${accountId}`;
  
  const [dailyCount, minuteCount] = await Promise.all([
    redis.incr(dailyKey),
    redis.incr(minuteKey),
  ]);
  
  if (dailyCount === 1) await redis.expire(dailyKey, 86400);
  if (minuteCount === 1) await redis.pexpire(minuteKey, 60000);
  
  const allowed = dailyCount <= limit.daily && minuteCount <= limit.perMinute;
  const remaining = Math.max(0, limit.daily - dailyCount);
  
  return { allowed, remaining };
}

---

5. Review Monitoring & Replies#

Daily Review Poll#

// src/server/bullmq/processors/review-monitor.ts

export async function reviewMonitorProcessor(job: Job) {
  const { practiceId } = job.data;
  
  const gbpAccounts = await db.gbpAccount.findMany({
    where: { practiceId, isActive: true },
    include: { locations: true },
  });
  
  for (const account of gbpAccounts) {
    for (const location of account.locations) {
      // Fetch new reviews
      const reviews = await fetchNewReviews(account.id, location.gbpLocationId);
      
      for (const review of reviews) {
        // Store review
        await db.review.create({
          data: {
            locationId: location.locationId,
            gbpLocationId: location.id,
            gbpReviewId: review.reviewId,
            reviewerName: reviewerName,
            rating: review.starRating,
            comment: review.comment,
            reviewDate: new Date(review.createTime),
            status: "NEW",
          },
        });
        
        // Generate AI reply
        const reply = await ai.generate({
          task: "review_reply",
          prompt: `Review: ${review.comment}\nRating: ${review.starRating}/5\nBusiness: ${location.name}`,
        });
        
        // Sentiment-based routing
        if (review.starRating >= 4) {
          // Positive: auto-reply
          await replyToReview(account.id, location.gbpLocationId, review.reviewId, reply.text);
          
          await db.review.update({
            where: { gbpReviewId: review.reviewId },
            data: {
              replyText: reply.text,
              replyGeneratedByAI: true,
              replyPublished: true,
              repliedAt: new Date(),
              status: "REPLIED",
            },
          });
        } else {
          // Negative: queue for approval
          await db.review.update({
            where: { gbpReviewId: review.reviewId },
            data: {
              replyText: reply.text,
              replyGeneratedByAI: true,
              replyPublished: false,
              status: "NEW", // Awaits approval
            },
          });
          
          // Notify client
          await email.send({
            to: practice.owner.email,
            subject: "New negative review requires your attention",
            body: `A ${review.starRating}-star review was received. Please approve the AI-generated reply in your dashboard.`,
          });
        }
      }
    }
  }
}

Review Reply Approval Flow#

---

6. Social Media Auth (Composio)#

Connection Flow#

// src/server/services/social/composio.ts

import { Composio } from "composio-core";

const composio = new Composio({ apiKey: process.env.COMPOSIO_API_KEY });

export async function initiateConnection(
  platform: string,
  redirectUri: string,
  metadata: Record<string, any>
) {
  const connection = await composio.connectedAccounts.initiate({
    appName: platform,
    redirectUri,
    metadata,
  });
  
  return {
    authUrl: connection.redirectUrl,
    connectionId: connection.connectedAccountId,
  };
}

export async function handleCallback(connectionId: string) {
  const connection = await composio.connectedAccounts.get({
    connectedAccountId: connectionId,
  });
  
  return {
    status: connection.status,
    appName: connection.appName,
    entityId: connection.entityId,
  };
}

Supported Platforms#

Social Account Storage#

model SocialAccount {
  id              String   @id @default(cuid())
  practiceId      String
  platform        PlatformType
  accountName     String
  accountId       String?
  profileUrl      String?
  accessToken     String   @db.Text
  refreshToken    String?  @db.Text
  tokenExpiresAt  DateTime?
  composioConnectionId String?
  followerCount   Int?
  isActive        Boolean  @default(true)
  lastSyncedAt    DateTime?
  
  practice        Practice @relation(fields: [practiceId], references: [id], onDelete: Cascade)
  posts           SocialPost[]
  
  @@unique([practiceId, platform])
  @@map("social_accounts")
}

---

7. Social Posting Pipeline#

Post Creation#

// src/server/services/social/platforms.ts

export async function publishPost(
  account: SocialAccount,
  content: string,
  mediaUrls?: string[]
): Promise<{ postId: string; url: string }> {
  switch (account.platform) {
    case "FACEBOOK":
      return publishFacebookPost(account.composioConnectionId!, content, mediaUrls);
    case "INSTAGRAM":
      return publishInstagramPost(account.composioConnectionId!, content, mediaUrls?.[0]);
    case "LINKEDIN":
      return publishLinkedInPost(account.composioConnectionId!, content);
    case "TWITTER":
      return publishTwitterPost(account.composioConnectionId!, content, mediaUrls);
    default:
      throw new Error(`Unsupported platform: ${account.platform}`);
  }
}

Facebook Post#

async function publishFacebookPost(
  connectionId: string,
  message: string,
  mediaUrls?: string[]
) {
  const action = await composio.getAction({ actionName: "FACEBOOK_PUBLISH_POST" });
  
  return await action.execute({
    connectedAccountId: connectionId,
    input: {
      message,
      ...(mediaUrls ? { photos: mediaUrls } : {}),
    },
  });
}

Instagram Post#

async function publishInstagramPost(
  connectionId: string,
  caption: string,
  imageUrl?: string
) {
  const action = await composio.getAction({ actionName: "INSTAGRAM_PUBLISH_POST" });
  
  return await action.execute({
    connectedAccountId: connectionId,
    input: {
      caption,
      image_url: imageUrl,
    },
  });
}

---

8. Zernio Scheduling#

Batch Scheduling#

// src/server/services/social/zernio.ts

export async function schedulePosts(posts: Array<{
  platform: string;
  connectionId: string;
  content: string;
  mediaUrls?: string[];
  scheduledFor: string;
}>) {
  const response = await fetch("https://api.zernio.com/v1/schedule/batch", {
    method: "POST",
    headers: {
      Authorization: `Bearer ${process.env.ZERNIO_API_KEY}`,
      "Content-Type": "application/json",
    },
    body: JSON.stringify({
      posts,
      webhook_url: `${process.env.APP_URL}/api/webhooks/zernio`,
    }),
  });
  
  if (!response.ok) {
    throw new Error(`Zernio scheduling failed: ${await response.text()}`);
  }
  
  return response.json();
}

Webhook Handler#

// src/app/api/webhooks/zernio/route.ts

export async function POST(request: Request) {
  const body = await request.json();
  
  // Update post status
  await db.socialPost.update({
    where: { id: body.postId },
    data: {
      status: body.status === "published" ? "PUBLISHED" : "FAILED",
      externalPostId: body.platformPostId,
      publishedAt: body.status === "published" ? new Date() : null,
      failedReason: body.error || null,
    },
  });
  
  return new Response("OK");
}

---

9. Content Approval Gate#

Approval States#

Medical Compliance Gate#

// All content for medical clients goes through approval
export async function publishWithApproval(
  content: ContentPiece,
  practice: Practice
) {
  if (practice.type === "CLINIC" || practice.type === "HOSPITAL") {
    // Medical clients: always require approval
    await db.contentPiece.update({
      where: { id: content.id },
      data: { status: "PENDING_REVIEW" },
    });
    
    // Schedule auto-publish in 24h
    await queues.emailSend.add("content-reminder", {
      contentId: content.id,
      practiceId: practice.id,
    }, { delay: 24 * 60 * 60 * 1000 });
    
    return { status: "PENDING_APPROVAL", autoPublishAt: new Date(Date.now() + 24 * 60 * 60 * 1000) };
  }
  
  // Non-medical: auto-publish
  return await publishImmediately(content);
}

---

10. Risk Mitigation#

GBP Suspension Prevention#

Emergency Stop#

// src/server/services/gbp/emergency-stop.ts

export async function emergencyStop(practiceId?: string) {
  if (practiceId) {
    // Stop one client
    await db.gbpAccount.updateMany({
      where: { practiceId },
      data: { isActive: false },
    });
  } else {
    // Stop ALL GBP activity (system-wide)
    await db.gbpAccount.updateMany({
      where: {},
      data: { isActive: false },
    });
  }
  
  // Cancel pending posts
  await queues.gbpPostPublish.pause();
  
  // Alert admin
  await email.send({
    to: "admin@rankflow.ai",
    subject: "🚨 GBP Emergency Stop Activated",
    body: `All GBP activity has been paused. Reason: ${reason}`,
  });
}

Social Platform Fallback#

// If Composio fails, queue for manual posting
export async function handleSocialFailure(
  post: SocialPost,
  error: Error
) {
  await db.socialPost.update({
    where: { id: post.id },
    data: {
      status: "FAILED",
      failedReason: error.message,
    },
  });
  
  // Notify client with copy-paste content
  await email.send({
    to: post.practice.owner.email,
    subject: "Your social post needs manual posting",
    body: `Platform: ${post.platform}\nContent: ${post.content}\nPlease post manually and mark as done in your dashboard.`,
  });
}

---

End of GBP & Social Media Pipeline Documentation