💳 Billing Completion Utility

📖 Overview

The billing.js utility handles the completion of Stripe billing data initialization. It monitors fetch progress, emits real-time notifications when initialization is complete, updates Stripe key status, and cleans up queue entries. This is a critical component of the billing setup flow.

Source File: queue-manager/common/billing.js

🎯 Purpose

• Progress Tracking: Monitors Stripe data fetch completion (6 fetch cycles required)
• User Notification: Real-time socket events when billing setup is complete
• Status Management: Updates Stripe key initialization status
• Queue Cleanup: Removes completed queue entries
• Initialization Handoff: Marks billing integration as ready for use

📘 Function Signature

completeProcess(token)

const completeProcess = async token => {
  // Implementation
};

Parameters

• token (Object) - Queue entry object containing:
  • id - Queue entry ID
  • user_id - User who initiated billing setup
  • account_id - Account being configured

Returns

• Promise<void> - Resolves when completion process finishes

🔧 Implementation Details

Complete Source Code

const StripeKey = require('../models/stripe-key');
const { socketEmit } = require('../utilities');
const Queue = require('../models/queues');

const completeProcess = async token => {
  const userId = token?.user_id;

  // Step 1: Fetch Stripe key with current fetch count
  const stripe_key = await StripeKey.findOne({ account_id: token?.account_id }).lean().exec();

  // Step 2: Check if initialization is complete (6 fetches)
  if (stripe_key.fetch_count >= 6) {
    // Step 3: Emit success notification to user
    await socketEmit('billing_data_fetched', [userId.toString()], {
      message: `Successfully fetched data, Please refresh the page`,
      type: 'success',
    });

    // Step 4: Reset fetch count and mark as initialized
    await StripeKey.updateOne(
      { account_id: token?.account_id },
      { $set: { fetch_count: 0, initialized: true } },
    );

    // Step 5: Cleanup queue entry
    await Queue.deleteOne({ _id: token.id });
  }
};

module.exports = {
  completeProcess,
};

📋 Step-by-Step Logic

Step 1: Fetch Stripe Key Document

const stripe_key = await StripeKey.findOne({ account_id: token?.account_id }).lean().exec();

Query Logic:

• Finds Stripe key configuration for the account
• Uses .lean() for performance (plain JavaScript object)
• .exec() explicitly executes the query

Stripe Key Document Structure:

{
  _id: ObjectId,
  account_id: ObjectId,
  stripe_publishable_key: 'pk_live_...',
  stripe_secret_key: 'sk_live_...',
  fetch_count: 0,              // Incremented by fetch processes
  initialized: false,          // Marks completion
  createdAt: Date,
  updatedAt: Date
}

Step 2: Check Completion Threshold

if (stripe_key.fetch_count >= 6) {
  // Complete initialization
}

Threshold Logic:

• fetch_count: Incremented by each fetch process (customers, products, prices, subscriptions, invoices, payment methods)
• 6 fetches required: Ensures all critical Stripe data is synchronized
• Greater than or equal: Handles edge cases where count exceeds 6

Why 6 Fetches?

Typical Stripe data synchronization requires fetching:

1. Customers - Customer records
2. Products - Product catalog
3. Prices - Pricing information
4. Subscriptions - Active subscriptions
5. Invoices - Invoice history
6. Payment Methods - Stored payment methods

Step 3: Socket Notification

await socketEmit('billing_data_fetched', [userId.toString()], {
  message: `Successfully fetched data, Please refresh the page`,
  type: 'success',
});

Notification Details:

• Event Name: 'billing_data_fetched'
• Recipients: Array with single user ID
• Payload:
  • message: User-facing success message
  • type: 'success' (for UI styling - green notification)

User Experience:

// Frontend socket listener
socket.on('billing_data_fetched', data => {
  if (data.type === 'success') {
    showNotification(data.message, 'success');
    // User refreshes to see synchronized billing data
  }
});

Why "Please refresh the page"?

• Billing data is cached in frontend state
• Refresh ensures latest synchronized data is displayed
• Alternative: Frontend could refetch data on notification

Step 4: Update Stripe Key Status

await StripeKey.updateOne(
  { account_id: token?.account_id },
  { $set: { fetch_count: 0, initialized: true } },
);

Update Operations:

1. Reset fetch_count to 0:

   • Clears the counter after successful initialization
   • Prepares for potential re-synchronization
   • Prevents counter from growing indefinitely

2. Set initialized to true:

   • Marks billing integration as ready
   • Frontend can enable billing features
   • Backend processes know setup is complete

Atomic Update: Uses updateOne for safe concurrent updates

Step 5: Queue Cleanup

await Queue.deleteOne({ _id: token.id });

Cleanup Logic:

• Removes completed queue entry
• Prevents duplicate processing
• Keeps queue collection clean
• No longer needed after initialization

Queue Entry Structure:

{
  _id: ObjectId,
  source: 'billing-initialization',
  status: 'processing',
  account_id: ObjectId,
  user_id: ObjectId,
  createdAt: Date
}

🔄 Complete Flow Diagram

sequenceDiagram
    participant FETCH as Fetch Processes
    participant STRIPE_KEY as StripeKey Model
    participant BILLING as billing.completeProcess()
    participant SOCKET as Socket.IO
    participant USER as User Browser
    participant QUEUE as Queue Collection

    FETCH->>STRIPE_KEY: Increment fetch_count (6 times)
    FETCH->>BILLING: Call completeProcess(token)

    BILLING->>STRIPE_KEY: Query fetch_count
    STRIPE_KEY-->>BILLING: fetch_count: 6

    alt fetch_count >= 6
        BILLING->>SOCKET: Emit 'billing_data_fetched'
        SOCKET->>USER: Success notification
        USER->>USER: Refresh page

        BILLING->>STRIPE_KEY: Update {fetch_count: 0, initialized: true}
        STRIPE_KEY-->>BILLING: Updated

        BILLING->>QUEUE: Delete queue entry
        QUEUE-->>BILLING: Deleted
    else fetch_count < 6
        BILLING->>BILLING: Skip (not ready)
    end

🎨 Usage Patterns

Typical Usage in Billing Module

const { completeProcess } = require('./common/billing');

// After each Stripe data fetch
async function fetchStripeCustomers(queueEntry) {
  try {
    // Fetch customers from Stripe
    const customers = await stripe.customers.list();

    // Save to database
    await saveCustomers(customers);

    // Increment fetch count
    await StripeKey.updateOne({ account_id: queueEntry.account_id }, { $inc: { fetch_count: 1 } });

    // Check if initialization is complete
    await completeProcess(queueEntry);
  } catch (error) {
    logger.error({ initiator: 'fetch-customers', error });
  }
}

Multi-Fetch Coordination

// Billing initialization service
async function initializeBilling(accountId, userId) {
  const queueEntry = await Queue.create({
    source: 'billing-initialization',
    account_id: accountId,
    user_id: userId,
  });

  // Start all fetch processes
  await Promise.all([
    fetchCustomers(queueEntry),
    fetchProducts(queueEntry),
    fetchPrices(queueEntry),
    fetchSubscriptions(queueEntry),
    fetchInvoices(queueEntry),
    fetchPaymentMethods(queueEntry),
  ]);

  // completeProcess() called by each fetch after incrementing count
}

📊 Data Structures

Token Parameter (Queue Entry)

{
    id: ObjectId('queue_entry_id'),
    user_id: ObjectId('user_id'),
    account_id: ObjectId('account_id'),
    source: 'billing-initialization',
    status: 'processing',
    createdAt: Date
}

StripeKey Document (Before Completion)

{
    _id: ObjectId,
    account_id: ObjectId,
    stripe_publishable_key: 'pk_live_abc123',
    stripe_secret_key: 'sk_live_xyz789',
    fetch_count: 5,              // Being incremented
    initialized: false,          // Not yet complete
    mode: 'live',
    createdAt: Date,
    updatedAt: Date
}

StripeKey Document (After Completion)

{
    _id: ObjectId,
    account_id: ObjectId,
    stripe_publishable_key: 'pk_live_abc123',
    stripe_secret_key: 'sk_live_xyz789',
    fetch_count: 0,              // Reset
    initialized: true,           // Marked complete ✅
    mode: 'live',
    createdAt: Date,
    updatedAt: Date
}

⚙️ Configuration

Required Models

const StripeKey = require('../models/stripe-key');
const Queue = require('../models/queues');

Socket Event

// Event name
'billing_data_fetched'

// Payload
{
    message: 'Successfully fetched data, Please refresh the page',
    type: 'success'
}

Fetch Count Threshold

const REQUIRED_FETCHES = 6;

// Can be made configurable
const threshold = process.env.BILLING_FETCH_THRESHOLD || 6;

🚨 Error Handling

Current Implementation

const completeProcess = async token => {
  // No try-catch wrapper
  // Errors propagate to caller
};

Error Behavior:

• Errors thrown to calling function
• Caller responsible for error handling
• No logging within this utility

Improved Error Handling

const completeProcess = async token => {
  try {
    const userId = token?.user_id;
    const stripe_key = await StripeKey.findOne({ account_id: token?.account_id }).lean().exec();

    if (!stripe_key) {
      logger.error({
        initiator: 'QM/billing-complete',
        message: 'Stripe key not found',
        account_id: token?.account_id,
      });
      return;
    }

    if (stripe_key.fetch_count >= 6) {
      await socketEmit('billing_data_fetched', [userId.toString()], {
        message: `Successfully fetched data, Please refresh the page`,
        type: 'success',
      });

      await StripeKey.updateOne(
        { account_id: token?.account_id },
        { $set: { fetch_count: 0, initialized: true } },
      );

      await Queue.deleteOne({ _id: token.id });
    }
  } catch (error) {
    logger.error({
      initiator: 'QM/billing-complete',
      error,
      account_id: token?.account_id,
    });
    throw error;
  }
};

📈 Performance Considerations

Optimization Strategies

1. Lean Query: Uses .lean() for faster reads
2. Conditional Logic: Only processes if threshold reached
3. Atomic Updates: Single updateOne operation
4. Non-blocking Socket: Socket emit doesn't wait for delivery

Scalability

• Single Account Processing: No cross-account locking
• Quick Operations: < 100ms typical execution
• Database Queries: 3 queries max (find, update, delete)
• Index Requirements:
  • StripeKey.account_id (unique index)
  • Queue._id (primary key)

🧪 Testing Considerations

Mock Setup

jest.mock('../models/stripe-key');
jest.mock('../models/queues');
jest.mock('../utilities', () => ({
  socketEmit: jest.fn(),
}));

const { completeProcess } = require('./common/billing');

Test Cases

describe('completeProcess', () => {
  test('Completes when fetch_count is 6', async () => {
    const token = {
      id: 'queue123',
      user_id: 'user123',
      account_id: 'account123',
    };

    StripeKey.findOne.mockReturnValue({
      lean: jest.fn().mockReturnThis(),
      exec: jest.fn().mockResolvedValue({
        fetch_count: 6,
        initialized: false,
      }),
    });

    await completeProcess(token);

    expect(socketEmit).toHaveBeenCalledWith(
      'billing_data_fetched',
      ['user123'],
      expect.objectContaining({ type: 'success' }),
    );

    expect(StripeKey.updateOne).toHaveBeenCalledWith(
      { account_id: 'account123' },
      { $set: { fetch_count: 0, initialized: true } },
    );

    expect(Queue.deleteOne).toHaveBeenCalledWith({ _id: 'queue123' });
  });

  test('Does nothing when fetch_count is less than 6', async () => {
    const token = {
      id: 'queue123',
      user_id: 'user123',
      account_id: 'account123',
    };

    StripeKey.findOne.mockReturnValue({
      lean: jest.fn().mockReturnThis(),
      exec: jest.fn().mockResolvedValue({
        fetch_count: 3,
        initialized: false,
      }),
    });

    await completeProcess(token);

    expect(socketEmit).not.toHaveBeenCalled();
    expect(StripeKey.updateOne).not.toHaveBeenCalled();
    expect(Queue.deleteOne).not.toHaveBeenCalled();
  });
});

🔗 Related Documentation

• Common Utilities Overview
• Billing Module
• Downgrade Logs - Similar socket notification pattern
• Stripe Integration (link removed - file does not exist) (if documented)

📝 Notes

Why 6 Fetches?

The magic number "6" represents the minimum Stripe data entities needed for billing functionality:

1. Customers
2. Products
3. Prices
4. Subscriptions
5. Invoices
6. Payment Methods

Initialization vs Synchronization

• Initialization: First-time setup when connecting Stripe
• Synchronization: Subsequent data updates (uses different process)
• This utility only handles initial setup completion

Alternative Approaches

Option 1: Single Fetch Process

// Instead of 6 separate fetches, could use single comprehensive fetch
// Pros: Simpler logic, single queue entry
// Cons: Longer processing time, all-or-nothing failure

Option 2: Progress Bar

// Could emit progress events (1/6, 2/6, etc.)
socketEmit('billing_progress', [userId], {
  progress: fetch_count,
  total: 6,
});
// Pros: Better UX with progress indication
// Cons: More socket events

Race Condition Consideration

If multiple fetch processes complete simultaneously:

• Each calls completeProcess()
• First one to check sees fetch_count: 6
• Updates and deletes queue entry
• Subsequent calls find deleted queue entry (silent failure acceptable)

---

Complexity: Low
Business Impact: High - Critical for billing setup
Dependencies: StripeKey model, Queue model, socketEmit utility
Last Updated: 2025-10-10