Marketer: for Digital Nomads A Deep Dive

// src/ingestion/kafka-consumer.js
// Core Kafka consumer for ingesting marketing events from nomad clients
// Handles event validation, retry logic, and dead-letter queue routing
const { Kafka, logLevel } = require('kafkajs');
const { z } = require('zod');
const { Redis } = require('@upstash/redis');
const { Client } = require('pg');
const pino = require('pino');

// Initialize dependencies
const logger = pino({ level: process.env.LOG_LEVEL || 'info' });
const redis = new Redis({
  url: process.env.REDIS_URL,
  token: process.env.REDIS_TOKEN,
});
const pgClient = new Client({ connectionString: process.env.POSTGRES_URL });
pgClient.connect().catch(err => logger.fatal({ err }, 'Failed to connect to PostgreSQL'));

// Event schema validation for marketing events
const MarketingEventSchema = z.object({
  eventId: z.string().uuid(),
  nomadId: z.string().uuid(),
  eventType: z.enum(['campaign_create', 'campaign_convert', 'compliance_check', 'payout_request']),
  payload: z.record(z.any()),
  timestamp: z.number().int().positive(),
  geolocation: z.object({
    countryCode: z.string().length(2),
    timezone: z.string(),
  }).optional(),
});

// Kafka configuration
const kafka = new Kafka({
  clientId: 'mfdn-ingestion-consumer',
  brokers: process.env.KAFKA_BROKERS.split(','),
  logLevel: logLevel.ERROR,
  retry: { initialRetryTime: 100, retries: 8 },
});

const consumer = kafka.consumer({ groupId: 'mfdn-marketing-events-v1' });

// Dead letter queue for invalid/failed events
const DLQ_TOPIC = 'mfdn-dlq-events';

// Graceful shutdown handler
const shutdown = async () => {
  logger.info('Initiating graceful shutdown...');
  await consumer.disconnect();
  await pgClient.end();
  redis.disconnect();
  logger.info('Shutdown complete');
  process.exit(0);
};
process.on('SIGTERM', shutdown);
process.on('SIGINT', shutdown);

// Main message processing loop
const run = async () => {
  await consumer.connect();
  logger.info('Kafka consumer connected, subscribing to topics');
  await consumer.subscribe({ topics: ['mfdn-marketing-events'], fromBeginning: false });

  await consumer.run({
    eachMessage: async ({ topic, partition, message }) => {
      const eventId = message.key?.toString() || 'unknown';
      try {
        // Parse and validate raw message
        const rawEvent = JSON.parse(message.value.toString());
        const validationResult = MarketingEventSchema.safeParse(rawEvent);
        if (!validationResult.success) {
          logger.warn({ eventId, errors: validationResult.error.errors }, 'Invalid event schema, routing to DLQ');
          await producer.send({
            topic: DLQ_TOPIC,
            messages: [{ key: eventId, value: JSON.stringify({ raw: rawEvent, errors: validationResult.error.errors }) }],
          });
          return;
        }
        const event = validationResult.data;

        // Enrich event with geolocation if missing
        if (!event.geolocation && event.payload.ip) {
          const geo = await redis.get(`geo:${event.payload.ip}`);
          event.geolocation = geo ? JSON.parse(geo) : { countryCode: 'XX', timezone: 'UTC' };
        }

        // Write hot data to Redis (1 hour TTL)
        await redis.set(`event:${event.eventId}`, JSON.stringify(event), { ex: 3600 });

        // Write cold data to PostgreSQL
        const query = `
          INSERT INTO marketing_events (event_id, nomad_id, event_type, payload, timestamp, geolocation, created_at)
          VALUES ($1, $2, $3, $4, $5, $6, NOW())
          ON CONFLICT (event_id) DO NOTHING;
        `;
        await pgClient.query(query, [
          event.eventId,
          event.nomadId,
          event.eventType,
          JSON.stringify(event.payload),
          new Date(event.timestamp),
          event.geolocation ? JSON.stringify(event.geolocation) : null,
        ]);

        logger.info({ eventId, eventType: event.eventType }, 'Event processed successfully');
      } catch (err) {
        logger.error({ err, eventId, topic, partition }, 'Failed to process message');
        // Retry 3 times before routing to DLQ
        const retryCount = await redis.incr(`retry:${eventId}`);
        if (retryCount <= 3) {
          await redis.expire(`retry:${eventId}`, 60);
          logger.info({ eventId, retryCount }, 'Retrying event processing');
          // Re-queue to end of topic (simplified for example)
          await producer.send({ topic, messages: [{ key: eventId, value: message.value }] });
        } else {
          logger.warn({ eventId }, 'Max retries exceeded, routing to DLQ');
          await producer.send({
            topic: DLQ_TOPIC,
            messages: [{ key: eventId, value: message.value.toString() }],
          });
          await redis.del(`retry:${eventId}`);
        }
      }
    },
  });
};

// Start consumer
run().catch(err => {
  logger.fatal({ err }, 'Failed to start Kafka consumer');
  process.exit(1);
});

// src/workflows/campaign-engine.js
// State machine for automating marketing campaigns across nomad geolocations
// Supports conditional triggers, budget caps, and cross-border compliance checks
const { Queue, Worker, Job } = require('bullmq');
const { Redis } = require('@upstash/redis');
const { Client } = require('pg');
const pino = require('pino');
const { z } = require('zod');

// Initialize dependencies
const logger = pino({ level: process.env.LOG_LEVEL || 'info' });
const redis = new Redis({ url: process.env.REDIS_URL, token: process.env.REDIS_TOKEN });
const pgClient = new Client({ connectionString: process.env.POSTGRES_URL });
pgClient.connect().catch(err => logger.fatal({ err }, 'PostgreSQL connection failed'));

// Workflow state schema
const WorkflowStateSchema = z.object({
  workflowId: z.string().uuid(),
  nomadId: z.string().uuid(),
  campaignId: z.string().uuid(),
  status: z.enum(['pending', 'running', 'paused', 'completed', 'failed']),
  currentStep: z.string(),
  steps: z.array(z.object({
    stepId: z.string(),
    type: z.enum(['send_email', 'post_social', 'check_compliance', 'adjust_budget']),
    config: z.record(z.any()),
    status: z.enum(['pending', 'running', 'completed', 'failed']),
    retryCount: z.number().int().min(0),
  })),
  budgetUsd: z.number().positive(),
  spentUsd: z.number().min(0),
  geofence: z.object({ countryCodes: z.array(z.string().length(2)), timezone: z.string() }),
  createdAt: z.date(),
  updatedAt: z.date(),
});

// BullMQ queue for workflow jobs
const workflowQueue = new Queue('mfdn-workflows', {
  connection: { url: process.env.REDIS_URL, token: process.env.REDIS_TOKEN },
  defaultJobOptions: {
    attempts: 3,
    backoff: { type: 'exponential', delay: 1000 },
    removeOnComplete: true,
    removeOnFail: false,
  },
});

// Worker to process workflow steps
const worker = new Worker('mfdn-workflows', async (job) => {
  const { workflowId, stepId } = job.data;
  logger.info({ workflowId, stepId }, 'Processing workflow step');

  try {
    // Fetch current workflow state
    const { rows } = await pgClient.query(
      'SELECT * FROM workflow_states WHERE workflow_id = $1',
      [workflowId]
    );
    if (rows.length === 0) throw new Error(`Workflow ${workflowId} not found`);
    const rawState = rows[0];
    const state = WorkflowStateSchema.parse(rawState);

    // Check budget cap
    if (state.spentUsd >= state.budgetUsd) {
      logger.warn({ workflowId }, 'Budget cap reached, pausing workflow');
      await pgClient.query(
        'UPDATE workflow_states SET status = $1, updated_at = NOW() WHERE workflow_id = $2',
        ['paused', workflowId]
      );
      return { status: 'paused', reason: 'budget_cap' };
    }

    // Find current step
    const step = state.steps.find(s => s.stepId === stepId);
    if (!step) throw new Error(`Step ${stepId} not found in workflow ${workflowId}`);
    if (step.status !== 'pending') {
      logger.info({ workflowId, stepId }, 'Step already processed, skipping');
      return { status: 'skipped' };
    }

    // Update step status to running
    await pgClient.query(
      'UPDATE workflow_steps SET status = $1, updated_at = NOW() WHERE step_id = $2',
      ['running', stepId]
    );

    // Execute step based on type
    let stepResult;
    switch (step.type) {
      case 'send_email':
        stepResult = await sendEmailStep(step.config, state.nomadId);
        break;
      case 'post_social':
        stepResult = await postSocialStep(step.config, state.geofence);
        break;
      case 'check_compliance':
        stepResult = await checkComplianceStep(step.config, state.geofence.countryCodes);
        break;
      case 'adjust_budget':
        stepResult = await adjustBudgetStep(step.config, state.budgetUsd, state.spentUsd);
        break;
      default:
        throw new Error(`Unsupported step type: ${step.type}`);
    }

    // Update step status to completed
    await pgClient.query(
      'UPDATE workflow_steps SET status = $1, result = $2, updated_at = NOW() WHERE step_id = $3',
      ['completed', JSON.stringify(stepResult), stepId]
    );

    // Check if all steps are completed
    const { rows: stepRows } = await pgClient.query(
      'SELECT COUNT(*) FROM workflow_steps WHERE workflow_id = $1 AND status != $2',
      [workflowId, 'completed']
    );
    if (parseInt(stepRows[0].count) === 0) {
      await pgClient.query(
        'UPDATE workflow_states SET status = $1, updated_at = NOW() WHERE workflow_id = $2',
        ['completed', workflowId]
      );
      logger.info({ workflowId }, 'Workflow completed successfully');
    }

    return { status: 'completed', stepResult };
  } catch (err) {
    logger.error({ err, workflowId, stepId }, 'Workflow step failed');
    // Increment retry count for step
    await pgClient.query(
      'UPDATE workflow_steps SET retry_count = retry_count + 1, status = $1 WHERE step_id = $2',
      ['failed', stepId]
    );
    // If max retries exceeded, fail workflow
    const { rows: retryRows } = await pgClient.query(
      'SELECT retry_count FROM workflow_steps WHERE step_id = $1',
      [stepId]
    );
    if (retryRows[0].retry_count >= 3) {
      await pgClient.query(
        'UPDATE workflow_states SET status = $1, updated_at = NOW() WHERE workflow_id = $2',
        ['failed', workflowId]
      );
      logger.error({ workflowId }, 'Max retries exceeded, workflow failed');
    }
    throw err; // Let BullMQ handle retry
  }
}, { connection: { url: process.env.REDIS_URL, token: process.env.REDIS_TOKEN } });

// Helper functions for step types
async function sendEmailStep(config, nomadId) {
  // Integrate with SendGrid/Mailgun here
  logger.info({ nomadId, template: config.templateId }, 'Sending email');
  return { sent: true, messageId: crypto.randomUUID() };
}

async function postSocialStep(config, geofence) {
  // Integrate with Twitter/Instagram/LinkedIn APIs here
  logger.info({ platforms: config.platforms, geofence }, 'Posting to social media');
  return { posted: true, postIds: config.platforms.map(() => crypto.randomUUID()) };
}

async function checkComplianceStep(config, countryCodes) {
  // Check GDPR, CCPA, local tax laws for target countries
  logger.info({ countryCodes }, 'Checking compliance');
  return { compliant: true, warnings: [] };
}

async function adjustBudgetStep(config, budget, spent) {
  const newBudget = spent + config.incrementUsd;
  logger.info({ oldBudget: budget, newBudget }, 'Adjusting budget');
  return { newBudget, adjusted: true };
}

// Graceful shutdown
worker.on('completed', job => logger.info({ jobId: job.id }, 'Job completed'));
worker.on('failed', (job, err) => logger.error({ jobId: job?.id, err }, 'Job failed'));
process.on('SIGTERM', async () => {
  await worker.close();
  await pgClient.end();
  redis.disconnect();
});

// src/api/graphql-server.js
// GraphQL API for serving aggregated marketing metrics to nomad dashboards
// Supports filtering by geolocation, time range, and campaign type
const { ApolloServer } = require('@apollo/server');
const { expressMiddleware } = require('@apollo/server/express4');
const { ApolloServerPluginDrainHttpServer } = require('@apollo/server/plugin/drainHttpServer');
const express = require('express');
const http = require('http');
const { Client } = require('pg');
const { Redis } = require('@upstash/redis');
const pino = require('pino');
const { z } = require('zod');

// Initialize dependencies
const logger = pino({ level: process.env.LOG_LEVEL || 'info' });
const pgClient = new Client({ connectionString: process.env.POSTGRES_URL });
const redis = new Redis({ url: process.env.REDIS_URL, token: process.env.REDIS_TOKEN });
const app = express();
const httpServer = http.createServer(app);

// GraphQL schema definition
const typeDefs = `#graphql
  type MarketingEvent {
    eventId: ID!
    nomadId: ID!
    eventType: EventType!
    payload: JSON!
    timestamp: DateTime!
    geolocation: Geolocation
    createdAt: DateTime!
  }

  type Geolocation {
    countryCode: String!
    timezone: String!
  }

  type CampaignMetrics {
    campaignId: ID!
    totalSpend: Float!
    conversions: Int!
    cpa: Float!
    reach: Int!
    topCountries: [CountryMetric!]!
  }

  type CountryMetric {
    countryCode: String!
    spend: Float!
    conversions: Int!
  }

  enum EventType {
    CAMPAIGN_CREATE
    CAMPAIGN_CONVERT
    COMPLIANCE_CHECK
    PAYOUT_REQUEST
  }

  input MetricsFilter {
    nomadId: ID
    campaignId: ID
    countryCodes: [String!]
    startTime: DateTime!
    endTime: DateTime!
  }

  type Query {
    getCampaignMetrics(filter: MetricsFilter!): CampaignMetrics!
    getRecentEvents(nomadId: ID!, limit: Int = 10): [MarketingEvent!]!
  }

  scalar DateTime
  scalar JSON
`;

// Custom scalar resolvers
const DateTime = {
  serialize(value) {
    return value instanceof Date ? value.toISOString() : value;
  },
  parseValue(value) {
    return new Date(value);
  },
  parseLiteral(ast) {
    if (ast.kind === 'StringValue') return new Date(ast.value);
    return null;
  },
};

const JSON = {
  serialize(value) {
    return value;
  },
  parseValue(value) {
    return value;
  },
  parseLiteral(ast) {
    if (ast.kind === 'StringValue') return JSON.parse(ast.value);
    return null;
  },
};

// Resolvers
const resolvers = {
  DateTime,
  JSON,
  Query: {
    getCampaignMetrics: async (_, { filter }) => {
      const cacheKey = `metrics:${JSON.stringify(filter)}`;
      // Check Redis cache first (5 minute TTL)
      const cached = await redis.get(cacheKey);
      if (cached) {
        logger.info({ filter }, 'Serving cached metrics');
        return JSON.parse(cached);
      }

      try {
        // Validate filter
        const filterSchema = z.object({
          nomadId: z.string().uuid().optional(),
          campaignId: z.string().uuid().optional(),
          countryCodes: z.array(z.string().length(2)).optional(),
          startTime: z.date(),
          endTime: z.date(),
        });
        const validatedFilter = filterSchema.parse(filter);

        // Build dynamic query
        let query = `
          SELECT 
            c.campaign_id,
            SUM(me.payload->>'spendUsd')::FLOAT AS total_spend,
            COUNT(DISTINCT me.event_id) FILTER (WHERE me.event_type = 'campaign_convert') AS conversions,
            c.reach,
            me.geolocation->>'countryCode' AS country_code,
            SUM(me.payload->>'spendUsd')::FLOAT AS country_spend,
            COUNT(DISTINCT me.event_id) FILTER (WHERE me.event_type = 'campaign_convert') AS country_conversions
          FROM marketing_events me
          JOIN campaigns c ON me.payload->>'campaignId' = c.campaign_id::TEXT
          WHERE me.timestamp BETWEEN $1 AND $2
        `;
        const params = [validatedFilter.startTime, validatedFilter.endTime];
        let paramIndex = 3;

        if (validatedFilter.nomadId) {
          query += ` AND me.nomad_id = $${paramIndex}`;
          params.push(validatedFilter.nomadId);
          paramIndex++;
        }
        if (validatedFilter.campaignId) {
          query += ` AND c.campaign_id = $${paramIndex}`;
          params.push(validatedFilter.campaignId);
          paramIndex++;
        }
        if (validatedFilter.countryCodes) {
          query += ` AND me.geolocation->>'countryCode' IN ($${paramIndex})`;
          params.push(validatedFilter.countryCodes);
          paramIndex++;
        }

        query += `
          GROUP BY c.campaign_id, c.reach, me.geolocation->>'countryCode'
          ORDER BY total_spend DESC
        `;

        const { rows } = await pgClient.query(query, params);
        if (rows.length === 0) throw new Error('No metrics found for filter');

        // Aggregate results
        const campaignId = rows[0].campaign_id;
        const totalSpend = rows.reduce((sum, row) => sum + parseFloat(row.total_spend), 0);
        const conversions = rows.reduce((sum, row) => sum + parseInt(row.conversions), 0);
        const reach = rows[0].reach;
        const cpa = conversions > 0 ? totalSpend / conversions : 0;
        const topCountries = rows.slice(0, 5).map(row => ({
          countryCode: row.country_code,
          spend: parseFloat(row.country_spend),
          conversions: parseInt(row.country_conversions),
        }));

        const result = {
          campaignId,
          totalSpend,
          conversions,
          cpa,
          reach,
          topCountries,
        };

        // Cache result
        await redis.set(cacheKey, JSON.stringify(result), { ex: 300 });
        return result;
      } catch (err) {
        logger.error({ err, filter }, 'Failed to fetch campaign metrics');
        throw new Error(`Metrics fetch failed: ${err.message}`);
      }
    },
    getRecentEvents: async (_, { nomadId, limit }) => {
      try {
        const { rows } = await pgClient.query(
          `SELECT * FROM marketing_events WHERE nomad_id = $1 ORDER BY timestamp DESC LIMIT $2`,
          [nomadId, limit]
        );
        return rows.map(row => ({
          eventId: row.event_id,
          nomadId: row.nomad_id,
          eventType: row.event_type.toUpperCase(),
          payload: row.payload,
          timestamp: row.timestamp,
          geolocation: row.geolocation,
          createdAt: row.created_at,
        }));
      } catch (err) {
        logger.error({ err, nomadId }, 'Failed to fetch recent events');
        throw new Error(`Recent events fetch failed: ${err.message}`);
      }
    },
  },
};

// Initialize Apollo Server
const server = new ApolloServer({
  typeDefs,
  resolvers,
  plugins: [ApolloServerPluginDrainHttpServer({ httpServer })],
});

// Start server
const startServer = async () => {
  await pgClient.connect();
  await server.start();
  app.use('/graphql', express.json(), expressMiddleware(server));
  await new Promise(resolve => httpServer.listen({ port: 4000 }, resolve));
  logger.info('GraphQL server running on http://localhost:4000/graphql');
};

startServer().catch(err => {
  logger.fatal({ err }, 'Failed to start GraphQL server');
  process.exit(1);
});

// Inject OpenTelemetry trace context into Kafka messages
const { trace } = require('@opentelemetry/api');
const { Kafka } = require('kafkajs');

const tracer = trace.getTracer('mfdn-producer');
async function sendEvent(event) {
  const span = tracer.startSpan('send-marketing-event');
  try {
    const ctx = trace.setSpanContext(span.spanContext());
    const headers = {};
    trace.inject(ctx, { setHeader: (key, value) => headers[key] = value });
    await kafka.producer.send({
      topic: 'mfdn-marketing-events',
      messages: [{
        key: event.eventId,
        value: JSON.stringify(event),
        headers: { ...headers, 'trace-id': span.spanContext().traceId },
      }],
    });
  } finally {
    span.end();
  }
}

// Precompute geolocation for IP ranges and cache in Redis
const { Reader } = require('@maxmind/geoip2-node');
const { Redis } = require('@upstash/redis');
const fs = require('fs');

const redis = new Redis({ url: process.env.REDIS_URL, token: process.env.REDIS_TOKEN });
const dbBuffer = fs.readFileSync('./GeoLite2-City.mmdb');
const reader = Reader.openBuffer(dbBuffer);

async function precomputeGeolocation(ip) {
  const cached = await redis.get(`geo:${ip}`);
  if (cached) return JSON.parse(cached);
  try {
    const geo = reader.city(ip);
    const result = {
      countryCode: geo.country.isoCode,
      timezone: geo.location.timeZone,
    };
    await redis.set(`geo:${ip}`, JSON.stringify(result), { ex: 604800 }); // 7 days
    return result;
  } catch (err) {
    return { countryCode: 'XX', timezone: 'UTC' };
  }
}

-- pg_partman configuration for marketing_events table
CREATE EXTENSION IF NOT EXISTS pg_partman;
SELECT partman.create_parent(
  p_parent_table => 'public.marketing_events',
  p_control => 'timestamp',
  p_type => 'time',
  p_interval => '1 month',
  p_premake => 1,
  p_start_partition => date_trunc('month', NOW())::TEXT
);
-- Retain 12 months of data
UPDATE partman.part_config SET retention = '12 months', retention_keep_table = false WHERE parent_table = 'public.marketing_events';