MoveNow

# MoveNow en local : docker compose up --build, puis http://localhost:8080
name: movenow-local

services:
  # L'émulateur Pub/Sub officiel de Google. Il remplace le vrai service, sans IAM ni BigQuery.
  pubsub:
    image: gcr.io/google.com/cloudsdktool/google-cloud-cli:emulators
    command: ["gcloud", "beta", "emulators", "pubsub", "start", "--project=movenow-local", "--host-port=0.0.0.0:8085"]
    healthcheck:
      test: ["CMD", "bash", "-c", "echo > /dev/tcp/127.0.0.1/8085"]
      interval: 3s
      retries: 40

  # Tableau de bord. En local, il imite aussi la BigQuery subscription et la dead-letter policy.
  tableau:
    build: ./tableau
    ports:
      - "8080:8080"
    environment:
      PUBSUB_EMULATOR_HOST: pubsub:8085
      PUBSUB_PROJECT_ID: movenow-local
      LOTS_DIR: /lots
    volumes:
      - lots:/lots
    depends_on:
      pubsub:
        condition: service_healthy

  # Le producteur publie en continu des lots d'une minute, avec deux messages invalides par lot.
  producteur:
    build: ./producteur
    command: ["--project", "movenow-local", "--topic", "positions", "--rate", "5", "--duration", "60",
              "--invalid", "2", "--batch", "lot-local", "--out", "/lots", "--loop"]
    environment:
      PUBSUB_EMULATOR_HOST: pubsub:8085
    volumes:
      - lots:/lots
    depends_on:
      tableau:
        condition: service_healthy

volumes:
  lots:

// Producteur de positions fictives pour MoveNow.
// Publie des lots cadencés sur un topic Pub/Sub et écrit, pour chaque lot, la liste des événements
// attendus dans <dossier>/<lot>.jsonl : c'est la base de la réconciliation avec BigQuery.
//
//   node producteur.mjs --project mon-projet --topic positions --rate 10 --duration 60 --batch lot-01
//   node producteur.mjs --topic positions --rate 5 --duration 10 --batch lot-02 --invalid 3
//   node producteur.mjs --rate 2 --duration 3 --dry-run           essai sans GCP
//
// Sur GCP, l'authentification passe par ADC. Si PUBSUB_EMULATOR_HOST est défini, la bibliothèque
// Pub/Sub publie vers l'émulateur local à la place.
import fs from 'node:fs';
import path from 'node:path';
import { parseArgs } from 'node:util';

const { values: options } = parseArgs({
  options: {
    topic: { type: 'string' },
    project: { type: 'string' },
    rate: { type: 'string', default: '10' },
    duration: { type: 'string', default: '60' },
    batch: { type: 'string', default: `lot-${new Date().toISOString().slice(0, 19).replace(/[-:T]/g, '')}` },
    invalid: { type: 'string', default: '0' },
    vehicles: { type: 'string', default: '20' },
    out: { type: 'string', default: '.' },
    loop: { type: 'boolean', default: false },
    'dry-run': { type: 'boolean', default: false },
  },
});

const rate = Number(options.rate);
const duration = Number(options.duration);
const invalid = Number(options.invalid);
const vehicles = Number(options.vehicles);
for (const [name, value, max] of [['rate', rate, 100], ['duration', duration, 600], ['invalid', invalid, 1000], ['vehicles', vehicles, 1000]]) {
  // Plafonds volontaires : ce script sert à des essais bornés, pas à reproduire le trafic métier.
  if (!Number.isInteger(value) || value < 0 || value > max) throw new Error(`--${name} doit être un entier entre 0 et ${max}`);
}
if (!/^[a-zA-Z0-9_-]+$/.test(options.batch)) throw new Error('--batch ne doit contenir que des lettres, chiffres, - et _');
if (!options['dry-run'] && !options.topic) throw new Error('--topic est requis, sauf avec --dry-run');
fs.mkdirSync(options.out, { recursive: true });

let topic = null;
if (!options['dry-run']) {
  const { PubSub } = await import('@google-cloud/pubsub');
  const projectId = options.project || process.env.GOOGLE_CLOUD_PROJECT || process.env.PUBSUB_PROJECT_ID;
  topic = new PubSub(projectId ? { projectId } : {}).topic(options.topic);
}

// Compteur global, écrit chaque seconde dans progression.json : en local, le tableau de bord s'en sert
// pour estimer le nombre de messages en attente.
let publiesTotal = 0;
const progression = (batch) => fs.writeFileSync(path.join(options.out, 'progression.json'), JSON.stringify({ lot: batch, publies: publiesTotal, maj: new Date().toISOString() }));

// Chaque véhicule part d'un point au hasard autour de Saint-Quentin, puis se déplace un peu à chaque position.
const flotte = Array.from({ length: vehicles }, () => ({ lat: 49.82 + Math.random() * 0.06, lon: 3.25 + Math.random() * 0.08 }));
function deplacer(n) {
  const v = flotte[n];
  v.lat = Math.min(49.88, Math.max(49.82, v.lat + (Math.random() - 0.5) * 0.002));
  v.lon = Math.min(3.33, Math.max(3.25, v.lon + (Math.random() - 0.5) * 0.003));
  return v;
}

async function publierLot(batch) {
  const total = rate * duration;
  // Les événements invalides sont répartis régulièrement dans le lot.
  const invalidCount = Math.min(invalid, total);
  const invalidAt = new Set(Array.from({ length: invalidCount }, (_, i) => 1 + Math.floor(((i + 0.5) * total) / invalidCount)));
  const records = [];
  const pending = [];
  const start = Date.now();
  let seq = 0;

  for (let second = 0; second < duration; second++) {
    for (let i = 0; i < rate; i++) {
      seq += 1;
      const n = seq % vehicles;
      const position = deplacer(n);
      const event = {
        event_id: `${batch}-${String(seq).padStart(6, '0')}`,
        vehicle_id: `vh-${String(n).padStart(4, '0')}`,
        event_time: new Date().toISOString(),
        latitude: Number(position.lat.toFixed(5)),
        longitude: Number(position.lon.toFixed(5)),
      };
      // Un message « invalide » est accepté par le topic mais ne respecte pas le schéma de la table :
      // sa latitude devient une chaîne de caractères.
      if (invalidAt.has(seq)) event.latitude = 'nord';
      const record = { event_id: event.event_id, invalide: invalidAt.has(seq), publie_le: event.event_time };
      records.push(record);
      publiesTotal += 1;
      if (topic) {
        pending.push(topic.publishMessage({ data: Buffer.from(JSON.stringify(event)) })
          .then((messageId) => { record.message_id = messageId; })
          .catch((error) => { record.erreur = error.message; }));
      } else if (seq <= 3) {
        console.log(JSON.stringify(event));
      }
    }
    if (topic) progression(batch);
    const wait = start + (second + 1) * 1000 - Date.now();
    if (wait > 0) await new Promise((resolve) => setTimeout(resolve, wait));
  }

  await Promise.all(pending);
  const file = path.join(options.out, `${batch}.jsonl`);
  fs.writeFileSync(file, records.map((record) => JSON.stringify(record)).join('\n') + '\n');
  const failed = records.filter((record) => record.erreur).length;
  console.log(`${batch} : ${records.length} événements, ${invalidAt.size} invalides, ${failed} échecs de publication${topic ? '' : ' (essai à blanc, rien publié)'}. Liste attendue : ${file}`);
  return failed;
}

if (options.loop) {
  // Mode continu pour la démonstration locale : un nouveau lot numéroté à la suite du précédent.
  for (let n = 1; ; n++) await publierLot(`${options.batch}-${String(n).padStart(3, '0')}`);
} else {
  process.exitCode = (await publierLot(options.batch)) ? 1 : 0;
}

// Sources de données du tableau de bord MoveNow. Trois modes :
//   demo        aucune variable : un flux simulé en mémoire, sans GCP ni Docker
//   emulateur   PUBSUB_EMULATOR_HOST défini : l'émulateur Pub/Sub local, avec une imitation
//               de la BigQuery subscription et de la dead-letter policy
//   bigquery    SOURCE=bigquery : lecture de la vraie table BigQuery sur GCP
import fs from 'node:fs';
import path from 'node:path';
import { bloc, blocCloudRun, blocIdentite, enCache } from './gcp.mjs';

const MAX_TENTATIVES = Number(process.env.MAX_DELIVERY_ATTEMPTS || 5);
const log = (severity, message, extra = {}) => console.log(JSON.stringify({ severity, message, ...extra }));

// Le contrat de la table : les mêmes contrôles que BigQuery quand la subscription écrit avec use_table_schema.
export function valider(data) {
  let event;
  try {
    event = JSON.parse(data);
  } catch {
    return { erreur: 'JSON illisible' };
  }
  if (!event || typeof event !== 'object') return { erreur: 'message vide' };
  if (typeof event.event_id !== 'string' || !event.event_id) return { erreur: 'event_id manquant' };
  if (typeof event.vehicle_id !== 'string' || !event.vehicle_id) return { erreur: 'vehicle_id manquant', event };
  if (typeof event.event_time !== 'string' || Number.isNaN(Date.parse(event.event_time))) return { erreur: 'event_time n’est pas un TIMESTAMP', event };
  for (const champ of ['latitude', 'longitude']) {
    if (typeof event[champ] !== 'number' || !Number.isFinite(event[champ])) return { erreur: `${champ} n’est pas un FLOAT`, event };
  }
  return { event };
}

const lireJson = (texte) => { try { return JSON.parse(texte); } catch { return null; } };

// Une « table » en mémoire, partagée par les modes demo et emulateur.
class TableLocale {
  constructor() {
    this.lignes = [];
    this.parId = new Map();
    this.derniers = new Map();
    this.deadLetter = [];
    this.idsDeadLetter = new Set();
    this.recus = 0;
    this.debit = [];
    this.depuisDernierPoint = 0;
    setInterval(() => {
      this.debit.push({ t: Date.now(), n: this.depuisDernierPoint });
      this.depuisDernierPoint = 0;
      if (this.debit.length > 60) this.debit.shift();
    }, 2000).unref();
  }
  ecrire(event) {
    const ligne = { ...event, ingere_le: new Date().toISOString() };
    this.lignes.push(ligne);
    if (this.lignes.length > 20000) this.lignes.splice(0, this.lignes.length - 20000);
    this.parId.set(event.event_id, (this.parId.get(event.event_id) || 0) + 1);
    this.derniers.set(event.vehicle_id, ligne);
    this.recus += 1;
    this.depuisDernierPoint += 1;
  }
  mettreDeCote(entree) {
    this.deadLetter.unshift(entree);
    if (this.deadLetter.length > 200) this.deadLetter.pop();
    if (entree.event_id) this.idsDeadLetter.add(entree.event_id);
  }
  positions() {
    return [...this.derniers.values()].map(({ vehicle_id, latitude, longitude, event_time }) => ({ vehicle_id, latitude, longitude, event_time }));
  }
  evenements() {
    return this.lignes.slice(-25).reverse();
  }
  lot(nom) {
    const prefixe = `${nom}-`;
    const recus = {};
    for (const [id, n] of this.parId) if (id.startsWith(prefixe)) recus[id] = n;
    return { recus, dead_letter: [...this.idsDeadLetter].filter((id) => id.startsWith(prefixe)) };
  }
  nomsDeLots() {
    const noms = new Set();
    for (const id of this.parId.keys()) noms.add(id.replace(/-\d{6}$/, ''));
    return [...noms].reverse().slice(0, 20);
  }
}

// Mode demo : un producteur et une imitation du transfert, tout en mémoire.
function sourceDemo() {
  const table = new TableLocale();
  const file = [];
  const lots = new Map();
  let actif = true;
  let publies = 0;
  let numeroLot = 0;
  let lot = null;
  const flotte = Array.from({ length: 20 }, () => ({ lat: 49.82 + Math.random() * 0.06, lon: 3.25 + Math.random() * 0.08 }));

  // Cinq événements par seconde, par lots d'une minute, avec un message invalide par lot.
  setInterval(() => {
    if (!lot || lot.seq >= 300) {
      numeroLot += 1;
      lot = { nom: `lot-demo-${String(numeroLot).padStart(3, '0')}`, seq: 0, attendus: [], invalide: 1 + Math.floor(Math.random() * 300) };
      lots.set(lot.nom, lot.attendus);
      if (lots.size > 10) lots.delete(lots.keys().next().value);
    }
    lot.seq += 1;
    const n = lot.seq % flotte.length;
    const v = flotte[n];
    v.lat = Math.min(49.88, Math.max(49.82, v.lat + (Math.random() - 0.5) * 0.002));
    v.lon = Math.min(3.33, Math.max(3.25, v.lon + (Math.random() - 0.5) * 0.003));
    const event = {
      event_id: `${lot.nom}-${String(lot.seq).padStart(6, '0')}`,
      vehicle_id: `vh-${String(n).padStart(4, '0')}`,
      event_time: new Date().toISOString(),
      latitude: Number(v.lat.toFixed(5)),
      longitude: Number(v.lon.toFixed(5)),
    };
    const invalide = lot.seq === lot.invalide;
    if (invalide) event.latitude = 'nord';
    lot.attendus.push({ event_id: event.event_id, invalide });
    file.push({ data: JSON.stringify(event), tentatives: 0 });
    publies += 1;
  }, 200).unref();

  // Le transfert traite au plus vingt messages toutes les 200 ms : le rattrapage reste visible.
  setInterval(() => {
    if (!actif) return;
    for (let i = 0; i < 20 && file.length; i++) {
      const message = file.shift();
      const resultat = valider(message.data);
      if (!resultat.erreur) { table.ecrire(resultat.event); continue; }
      message.tentatives += 1;
      if (message.tentatives < MAX_TENTATIVES) { file.push(message); continue; }
      table.mettreDeCote({ event_id: resultat.event?.event_id || null, erreur: resultat.erreur, tentatives: message.tentatives, recu_le: new Date().toISOString(), extrait: message.data.slice(0, 160) });
    }
  }, 200).unref();

  return {
    mode: 'demo',
    libelle: 'Démo en mémoire',
    pret: Promise.resolve(),
    async etat() {
      return { compteurs: { recus: table.recus, dead_letter: table.deadLetter.length, en_attente: file.length, publies }, debit: table.debit, pas_secondes: 2, transfert: { pilotable: true, actif } };
    },
    async positions() { return table.positions(); },
    async evenements() { return table.evenements(); },
    async deadLetter() { return table.deadLetter.slice(0, 30); },
    async lots() { return [...lots.keys()].reverse().map((nom) => ({ nom, attendus: lots.get(nom).length })); },
    async lot(nom) { return { nom, attendus: lots.get(nom) || null, ...table.lot(nom) }; },
    async piloter(etat) { actif = etat; return { actif }; },
    async deploiement() {
      return { plateforme: 'local', blocs: [
        bloc('producteur', 'Producteur', 'simule', 'producteur intégré au tableau, cinq positions par seconde'),
        bloc('topic', 'Topic Pub/Sub', 'simule', 'file en mémoire'),
        bloc('subscription', 'Transfert vers la table', 'simule', actif ? 'transfert imité, actif' : 'transfert imité, coupé'),
        bloc('deadletter', 'Dead-letter', 'simule', `${table.deadLetter.length} message(s) mis de côté après ${MAX_TENTATIVES} essais`),
        bloc('inspection', 'Subscription d’inspection', 'simule', 'liste affichée dans le tableau'),
        bloc('table', 'Table BigQuery', 'simule', `${table.recus} positions en mémoire`),
        blocCloudRun('Tableau de bord sur Cloud Run'),
        bloc('identite', 'Identité du tableau', 'simule', 'pas de compte de service en local'),
        bloc('obs', 'Monitoring et alertes', 'manuel', 'backlog, âge du plus ancien message et alertes : à montrer dans Monitoring'),
      ] };
    },
  };
}

// Mode emulateur : vrai Pub/Sub local. L'émulateur ne sait pas écrire dans BigQuery ni appliquer
// une dead-letter policy : ce module les imite, pour que le flux se comporte comme sur GCP.
async function sourceEmulateur(env) {
  const { PubSub } = await import('@google-cloud/pubsub');
  const pubsub = new PubSub({ projectId: env.PUBSUB_PROJECT_ID || 'movenow-local' });
  const noms = {
    topic: env.TOPIC || 'positions',
    subscription: env.SUBSCRIPTION || 'positions-vers-bigquery',
    deadLetterTopic: env.DEAD_LETTER_TOPIC || 'positions-dead-letter',
    inspection: env.DEAD_LETTER_SUBSCRIPTION || 'positions-dead-letter-inspection',
  };
  const dossierLots = env.LOTS_DIR || '/lots';
  const table = new TableLocale();
  const tentatives = new Map();
  let abonnement = null;

  // Crée les ressources si elles n'existent pas encore (code 6 : ALREADY_EXISTS).
  const assurer = async (creer) => { try { await creer(); } catch (error) { if (error.code !== 6) throw error; } };
  async function initialiser() {
    await assurer(() => pubsub.createTopic(noms.topic));
    await assurer(() => pubsub.createTopic(noms.deadLetterTopic));
    await assurer(() => pubsub.topic(noms.topic).createSubscription(noms.subscription, { ackDeadlineSeconds: 10 }));
    await assurer(() => pubsub.topic(noms.deadLetterTopic).createSubscription(noms.inspection));
    ouvrir();
    pubsub.subscription(noms.inspection).on('message', (message) => {
      const event = lireJson(message.data.toString());
      table.mettreDeCote({
        event_id: event?.event_id || null,
        erreur: message.attributes.erreur || 'inconnue',
        tentatives: Number(message.attributes.tentatives) || null,
        recu_le: new Date().toISOString(),
        extrait: message.data.toString().slice(0, 160),
      });
      message.ack();
    });
    log('INFO', 'topics et subscriptions prêts dans l’émulateur', noms);
  }
  // L'émulateur peut mettre quelques secondes à répondre : on réessaie pendant une minute.
  const pret = (async () => {
    for (let essai = 1; ; essai++) {
      try {
        return await initialiser();
      } catch (error) {
        if (essai >= 30) throw error;
        log('WARNING', `émulateur pas encore prêt : ${error.message}`);
        await new Promise((resolve) => setTimeout(resolve, 2000));
      }
    }
  })();

  function ouvrir() {
    const deadLetterTopic = pubsub.topic(noms.deadLetterTopic);
    abonnement = pubsub.subscription(noms.subscription, { flowControl: { maxMessages: 500 } });
    abonnement.on('message', async (message) => {
      const resultat = valider(message.data.toString());
      if (!resultat.erreur) {
        table.ecrire(resultat.event);
        tentatives.delete(message.id);
        message.ack();
        return;
      }
      const n = (tentatives.get(message.id) || 0) + 1;
      tentatives.set(message.id, n);
      if (n < MAX_TENTATIVES) { message.nack(); return; }
      tentatives.delete(message.id);
      await deadLetterTopic.publishMessage({ data: message.data, attributes: { erreur: resultat.erreur, tentatives: String(n), source: noms.subscription } });
      message.ack();
    });
    abonnement.on('error', (error) => log('ERROR', error.message, { code: error.code }));
  }

  function lireAttendus(nom) {
    const fichier = path.join(dossierLots, `${nom}.jsonl`);
    if (!/^[\w-]+$/.test(nom) || !fs.existsSync(fichier)) return null;
    return fs.readFileSync(fichier, 'utf8').trim().split('\n').map(lireJson).filter(Boolean).map(({ event_id, invalide }) => ({ event_id, invalide }));
  }

  return {
    mode: 'emulateur',
    libelle: 'Émulateur Pub/Sub local',
    pret,
    async etat() {
      const progression = fs.existsSync(path.join(dossierLots, 'progression.json')) ? lireJson(fs.readFileSync(path.join(dossierLots, 'progression.json'), 'utf8')) : null;
      const publies = progression?.publies ?? null;
      const enAttente = publies === null ? null : Math.max(0, publies - table.recus - table.deadLetter.length);
      return { compteurs: { recus: table.recus, dead_letter: table.deadLetter.length, en_attente: enAttente, publies }, debit: table.debit, pas_secondes: 2, transfert: { pilotable: true, actif: Boolean(abonnement) } };
    },
    async positions() { return table.positions(); },
    async evenements() { return table.evenements(); },
    async deadLetter() { return table.deadLetter.slice(0, 30); },
    async lots() {
      const fichiers = fs.existsSync(dossierLots) ? fs.readdirSync(dossierLots).filter((f) => f.endsWith('.jsonl')).map((f) => f.slice(0, -6)) : [];
      const noms = [...new Set([...table.nomsDeLots(), ...fichiers])].sort().reverse().slice(0, 20);
      return noms.map((nom) => ({ nom, attendus: lireAttendus(nom)?.length ?? null }));
    },
    async lot(nom) { return { nom, attendus: lireAttendus(nom), ...table.lot(nom) }; },
    async deploiement() {
      const progression = fs.existsSync(path.join(dossierLots, 'progression.json')) ? lireJson(fs.readFileSync(path.join(dossierLots, 'progression.json'), 'utf8')) : null;
      const recent = progression && Date.now() - Date.parse(progression.maj) < 10000;
      return { plateforme: 'local', blocs: [
        bloc('producteur', 'Producteur', 'simule', recent ? `conteneur producteur actif, lot ${progression.lot}` : 'aucune publication récente du conteneur producteur'),
        bloc('topic', 'Topic Pub/Sub', 'simule', `émulateur Pub/Sub, topic ${noms.topic}`),
        bloc('subscription', 'Transfert vers la table', 'simule', `subscription ${noms.subscription}, transfert imité ${abonnement ? 'actif' : 'coupé'}`),
        bloc('deadletter', 'Dead-letter', 'simule', `topic ${noms.deadLetterTopic}, ${MAX_TENTATIVES} essais imités`),
        bloc('inspection', 'Subscription d’inspection', 'simule', `${noms.inspection}, ${table.deadLetter.length} message(s) reçu(s)`),
        bloc('table', 'Table BigQuery', 'simule', `${table.recus} positions en mémoire`),
        blocCloudRun('Tableau de bord sur Cloud Run'),
        bloc('identite', 'Identité du tableau', 'simule', 'pas de compte de service en local'),
        bloc('obs', 'Monitoring et alertes', 'manuel', 'backlog, âge du plus ancien message et alertes : à montrer dans Monitoring'),
      ] };
    },
    // Couper le transfert imite le retrait du droit d'écriture : les messages s'accumulent dans Pub/Sub.
    async piloter(actif) {
      if (actif && !abonnement) ouvrir();
      if (!actif && abonnement) { await abonnement.close(); abonnement = null; }
      return { actif: Boolean(abonnement) };
    },
  };
}

// Mode bigquery : lecture seule de la table réelle, rafraîchie toutes les REFRESH_SECONDS secondes.
// Chaque rafraîchissement lance quelques petites requêtes filtrées sur la partition.
async function sourceBigQuery(env) {
  const { BigQuery } = await import('@google-cloud/bigquery');
  if (!/^[\w-]+\.[\w-]+\.[\w-]+$/.test(env.BQ_TABLE || '')) throw new Error('BQ_TABLE doit être de la forme projet.dataset.table');
  const bigquery = new BigQuery(env.GOOGLE_CLOUD_PROJECT ? { projectId: env.GOOGLE_CLOUD_PROJECT } : {});
  const table = `\`${env.BQ_TABLE}\``;
  const requete = async (query, params = {}) => (await bigquery.query({ query, params, location: env.BQ_LOCATION || undefined }))[0];
  const valeur = (v) => (v && typeof v === 'object' && 'value' in v ? v.value : v);
  const cache = { compteurs: { recus: 0, dead_letter: null, en_attente: null, publies: null }, debit: [], positions: [], evenements: [], deadLetter: [], erreur: null };

  let lireDeadLetter = async () => [];
  if (env.DEAD_LETTER_SUBSCRIPTION) {
    // Lecture sans acquittement : les messages sont aussitôt rendus à la subscription d'inspection.
    const { v1 } = await import('@google-cloud/pubsub');
    const client = new v1.SubscriberClient();
    const subscription = env.DEAD_LETTER_SUBSCRIPTION.includes('/') ? env.DEAD_LETTER_SUBSCRIPTION : `projects/${env.GOOGLE_CLOUD_PROJECT}/subscriptions/${env.DEAD_LETTER_SUBSCRIPTION}`;
    lireDeadLetter = async () => {
      const [reponse] = await client.pull({ subscription, maxMessages: 20 });
      const messages = reponse.receivedMessages || [];
      if (messages.length) await client.modifyAckDeadline({ subscription, ackIds: messages.map((m) => m.ackId), ackDeadlineSeconds: 0 });
      return messages.map(({ message }) => {
        const data = Buffer.from(message.data || '').toString();
        return {
          event_id: lireJson(data)?.event_id || null,
          erreur: 'voir les journaux de la subscription',
          tentatives: Number(message.attributes?.CloudPubSubDeadLetterSourceDeliveryCount) || null,
          recu_le: message.publishTime ? new Date(Number(message.publishTime.seconds) * 1000).toISOString() : null,
          extrait: data.slice(0, 160),
        };
      });
    };
  }

  async function rafraichir() {
    try {
      const [compte] = await requete(`SELECT COUNT(*) AS recus FROM ${table} WHERE event_time > TIMESTAMP_SUB(CURRENT_TIMESTAMP(), INTERVAL 1 HOUR)`);
      const debit = await requete(`SELECT UNIX_SECONDS(TIMESTAMP_TRUNC(event_time, MINUTE)) AS t, COUNT(*) AS n FROM ${table}
        WHERE event_time > TIMESTAMP_SUB(CURRENT_TIMESTAMP(), INTERVAL 30 MINUTE) GROUP BY t ORDER BY t`);
      const positions = await requete(`SELECT vehicle_id, latitude, longitude, event_time FROM ${table}
        WHERE event_time > TIMESTAMP_SUB(CURRENT_TIMESTAMP(), INTERVAL 10 MINUTE)
        QUALIFY ROW_NUMBER() OVER (PARTITION BY vehicle_id ORDER BY event_time DESC) = 1`);
      const evenements = await requete(`SELECT event_id, vehicle_id, event_time, latitude, longitude FROM ${table}
        WHERE event_time > TIMESTAMP_SUB(CURRENT_TIMESTAMP(), INTERVAL 1 HOUR) ORDER BY event_time DESC LIMIT 25`);
      try {
        cache.deadLetter = await lireDeadLetter();
        cache.erreurDeadLetter = null;
      } catch (error) {
        cache.erreurDeadLetter = error.message;
      }
      cache.compteurs = { recus: compte.recus, dead_letter: env.DEAD_LETTER_SUBSCRIPTION ? cache.deadLetter.length : null, en_attente: null, publies: null };
      cache.debit = debit.map(({ t, n }) => ({ t: t * 1000, n }));
      cache.positions = positions.map((p) => ({ ...p, event_time: valeur(p.event_time) }));
      cache.evenements = evenements.map((e) => ({ ...e, event_time: valeur(e.event_time) }));
      cache.erreur = null;
    } catch (error) {
      cache.erreur = error.message;
      log('ERROR', error.message);
    }
  }
  const pret = rafraichir();
  setInterval(rafraichir, Number(env.REFRESH_SECONDS || 20) * 1000).unref();

  return {
    mode: 'bigquery',
    libelle: `BigQuery, ${env.BQ_TABLE}`,
    pret,
    async etat() { return { compteurs: cache.compteurs, debit: cache.debit, transfert: { pilotable: false, actif: null }, erreur: cache.erreur, pas_secondes: 60 }; },
    async positions() { return cache.positions; },
    async evenements() { return cache.evenements; },
    async deadLetter() { return cache.deadLetter; },
    async lots() {
      const lignes = await requete(`SELECT REGEXP_EXTRACT(event_id, r'^(.*)-[0-9]{6}$') AS nom, COUNT(*) AS n FROM ${table}
        WHERE event_time > TIMESTAMP_SUB(CURRENT_TIMESTAMP(), INTERVAL 1 DAY) GROUP BY nom ORDER BY MAX(event_time) DESC LIMIT 20`);
      return lignes.filter((l) => l.nom).map((l) => ({ nom: l.nom, attendus: null }));
    },
    async lot(nom) {
      const lignes = await requete(`SELECT event_id, COUNT(*) AS n FROM ${table}
        WHERE STARTS_WITH(event_id, @prefixe) AND event_time > TIMESTAMP_SUB(CURRENT_TIMESTAMP(), INTERVAL 3 DAY) GROUP BY event_id`, { prefixe: `${nom}-` });
      return { nom, attendus: null, recus: Object.fromEntries(lignes.map((l) => [l.event_id, l.n])), dead_letter: cache.deadLetter.map((d) => d.event_id).filter((id) => id?.startsWith(`${nom}-`)) };
    },
    async piloter() { throw Object.assign(new Error('sur GCP, coupez le transfert en retirant le droit d’écriture du service agent'), { http: 409 }); },
    deploiement: enCache(() => carteBigQuery(env, bigquery, cache), 20),
  };
}

// Carte du déploiement sur GCP. La table se lit avec bigquery.dataViewer, déjà nécessaire au tableau.
// La configuration de la subscription demande roles/pubsub.viewer sur elle, ce qui reste facultatif.
const contrat = { event_id: 'STRING', vehicle_id: 'STRING', event_time: 'TIMESTAMP', latitude: 'FLOAT', longitude: 'FLOAT' };
async function carteBigQuery(env, bigquery, cache) {
  const blocs = [blocCloudRun('Tableau de bord sur Cloud Run'), await blocIdentite()];
  const recus = cache.compteurs.recus;
  blocs.push(recus > 0
    ? bloc('producteur', 'Producteur', 'ok', `${recus} position(s) écrite(s) dans la dernière heure`, 'requête sur la table')
    : bloc('producteur', 'Producteur', 'inconnu', 'aucune position récente dans la table : producteur lancé, transfert configuré ?'));

  const [projet, dataset, nomTable] = env.BQ_TABLE.split('.');
  try {
    const [meta] = await bigquery.dataset(dataset, { projectId: projet }).table(nomTable).getMetadata();
    const champs = Object.fromEntries((meta.schema?.fields || []).map((f) => [f.name, f.type === 'FLOAT64' ? 'FLOAT' : f.type]));
    const ecarts = Object.entries(contrat).filter(([nom, type]) => champs[nom] !== type).map(([nom, type]) => `${nom} : ${champs[nom] || 'absent'} au lieu de ${type}`);
    const partition = meta.timePartitioning?.field ? `partitionnée sur ${meta.timePartitioning.field}` : meta.timePartitioning ? 'partitionnée par date d’ingestion' : 'non partitionnée';
    const filtre = meta.requirePartitionFilter ? ', filtre de partition exigé' : '';
    blocs.push(ecarts.length
      ? bloc('table', 'Table BigQuery', 'alerte', `schéma différent du contrat du producteur : ${ecarts.join(', ')}`, env.BQ_TABLE)
      : bloc('table', 'Table BigQuery', 'ok', `${partition}${filtre}, schéma conforme au contrat du producteur`, env.BQ_TABLE));
  } catch (error) {
    blocs.push(bloc('table', 'Table BigQuery', 'echec', 'métadonnées de la table illisibles', error.message));
  }

  if (!env.SUBSCRIPTION) {
    blocs.push(bloc('topic', 'Topic Pub/Sub', 'inconnu', 'renseignez SUBSCRIPTION pour afficher ce bloc'), bloc('subscription', 'BigQuery subscription', 'inconnu', 'renseignez SUBSCRIPTION pour afficher ce bloc'), bloc('deadletter', 'Dead-letter', 'inconnu', 'renseignez SUBSCRIPTION pour afficher ce bloc'));
  } else {
    try {
      const { PubSub } = await import('@google-cloud/pubsub');
      const [sub] = await new PubSub(env.GOOGLE_CLOUD_PROJECT ? { projectId: env.GOOGLE_CLOUD_PROJECT } : {}).subscription(env.SUBSCRIPTION).getMetadata();
      const court = (nom = '') => nom.split('/').pop();
      blocs.push(bloc('topic', 'Topic Pub/Sub', 'ok', `topic ${court(sub.topic)}`, sub.topic));
      blocs.push(sub.bigqueryConfig?.table
        ? bloc('subscription', 'BigQuery subscription', 'ok', `écrit dans ${sub.bigqueryConfig.table}, use_table_schema ${sub.bigqueryConfig.useTableSchema ? 'oui' : 'non'}`, court(sub.name))
        : bloc('subscription', 'Transfert vers la table', 'ok', 'subscription sans export BigQuery : un consommateur fait le transfert', court(sub.name)));
      blocs.push(sub.deadLetterPolicy?.deadLetterTopic
        ? bloc('deadletter', 'Dead-letter', 'ok', `vers ${court(sub.deadLetterPolicy.deadLetterTopic)}, ${sub.deadLetterPolicy.maxDeliveryAttempts} tentatives au plus`, 'deadLetterPolicy')
        : bloc('deadletter', 'Dead-letter', 'inconnu', 'pas de dead-letter policy sur la subscription'));
    } catch (error) {
      const droit = error.code === 7 ? 'pour afficher ce bloc, donnez au tableau roles/pubsub.viewer sur la subscription (facultatif)' : 'configuration de la subscription illisible';
      blocs.push(bloc('topic', 'Topic Pub/Sub', 'inconnu', droit), bloc('subscription', 'BigQuery subscription', 'inconnu', droit, error.message), bloc('deadletter', 'Dead-letter', 'inconnu', droit));
    }
  }

  blocs.push(!env.DEAD_LETTER_SUBSCRIPTION
    ? bloc('inspection', 'Subscription d’inspection', 'inconnu', 'renseignez DEAD_LETTER_SUBSCRIPTION pour afficher ce bloc')
    : cache.erreurDeadLetter
      ? bloc('inspection', 'Subscription d’inspection', 'inconnu', 'lecture impossible : roles/pubsub.subscriber sur la subscription d’inspection ?', cache.erreurDeadLetter)
      : bloc('inspection', 'Subscription d’inspection', 'ok', `lisible, ${cache.deadLetter.length} message(s) visible(s)`, env.DEAD_LETTER_SUBSCRIPTION));
  blocs.push(bloc('obs', 'Monitoring et alertes', 'manuel', 'backlog, âge du plus ancien message et alertes : à montrer dans Monitoring'));
  return { plateforme: process.env.K_SERVICE ? 'Cloud Run' : 'local', blocs };
}

export async function creerSource(env = process.env) {
  if (env.SOURCE === 'bigquery') return sourceBigQuery(env);
  if (env.PUBSUB_EMULATOR_HOST) return sourceEmulateur(env);
  return sourceDemo();
}

# MoveNow, kit de démarrage

Un producteur de positions fictives et un tableau de bord qui montre ce que le pipeline a vraiment écrit : la carte des véhicules, le débit, la dead-letter et la réconciliation des lots. Le kit est un démonstrateur : il sert à vérifier votre infrastructure, il n’est pas évalué.

## Ce qu’il contient

```text
producteur/           Publie des lots cadencés sur Pub/Sub et écrit la liste des événements attendus
tableau/              Tableau de bord, déployable sur Cloud Run en lecture seule
  server.mjs          Routes de l’API et de l’interface
  sources.mjs         Démo en mémoire, émulateur Pub/Sub local ou BigQuery réel
  public/             Interface : carte, débit, flux, dead-letter, réconciliation
docker-compose.yml    Émulateur Pub/Sub, tableau et producteur en continu
```

## Lancer en local

Le plus rapide, sans Docker, sans GCP et sans rien installer : un flux simulé entièrement en mémoire.

```sh
cd tableau
node server.mjs
```

Avec Docker, le flux passe par le vrai émulateur Pub/Sub de Google. Le premier lancement télécharge son image, qui est volumineuse.

```sh
docker compose up --build
```

Ouvrez ensuite http://localhost:8080. Les véhicules bougent sur la carte, quelques messages invalides partent en dead-letter après cinq tentatives. Le bouton « Couper le transfert » imite le retrait du droit d’écriture : la file d’attente grossit, puis se résorbe quand vous rétablissez.

| Rôle | En local | Sur GCP |
| --- | --- | --- |
| Topic et subscriptions | émulateur Pub/Sub | Pub/Sub |
| Transfert vers la table | imité par le tableau (`sources.mjs`) | BigQuery subscription |
| Table analytique | mémoire du tableau | table BigQuery partitionnée sur `event_time` |
| Dead-letter | imitée : cinq essais, puis publication sur le topic d’échec | dead-letter policy de la subscription |
| Listes attendues | volume partagé `lots` | fichiers `.jsonl` à charger dans l’interface |

L’émulateur ne connaît ni IAM, ni les BigQuery subscriptions, ni les métriques de Monitoring. Ce que vous observez en local montre le principe, pas le comportement exact de GCP.

## Passer sur GCP

### Le producteur

Il tourne depuis votre poste ou Cloud Shell, avec ADC. Son identité n’a besoin que de publier sur le topic.

```sh
cd producteur
npm install
node producteur.mjs --project votre-projet --topic positions --rate 1 --duration 10 --batch lot-01
node producteur.mjs --project votre-projet --topic positions --rate 5 --duration 10 --batch lot-02 --invalid 3
```

Chaque lot laisse un fichier `lot-XX.jsonl` : la liste des identifiants attendus. Le débit est plafonné à 100 événements par seconde et la durée à 10 minutes par lot.

### Le contrat de la table

Le producteur envoie du JSON avec ces champs, qui doivent correspondre au schéma de la table quand la subscription écrit avec `use_table_schema` :

| Champ | Type |
| --- | --- |
| `event_id` | STRING |
| `vehicle_id` | STRING |
| `event_time` | TIMESTAMP |
| `latitude` | FLOAT |
| `longitude` | FLOAT |

L’option `--invalid` remplace la latitude par du texte : le topic accepte le message, la table le refuse.

### Le tableau de bord

Il se déploie sur Cloud Run et lit la table en lecture seule. Son identité a besoin de lire la table et de lancer des requêtes, et de s’abonner à la subscription d’inspection si vous voulez voir la dead-letter.

```sh
REGION=europe-west9
PROJECT_ID=votre-projet
IMAGE="$REGION-docker.pkg.dev/$PROJECT_ID/movenow/tableau:v1"
gcloud builds submit --tag "$IMAGE" tableau
```

| Variable | Rôle |
| --- | --- |
| `SOURCE` | `bigquery` pour lire la vraie table. |
| `BQ_TABLE` | `projet.dataset.table`. |
| `BQ_LOCATION` | La localisation du dataset, par exemple `europe-west9`. |
| `GOOGLE_CLOUD_PROJECT` | Le projet, utile pour les noms courts. |
| `DEAD_LETTER_SUBSCRIPTION` | Facultatif. La subscription d’inspection, lue sans acquitter les messages. |
| `REFRESH_SECONDS` | Intervalle de rafraîchissement, 20 secondes par défaut. |

Chaque rafraîchissement lance quelques petites requêtes filtrées sur la partition. C’est peu, mais pas gratuit : fermez le tableau de bord après la démonstration. Le nombre de messages en attente ne se lit pas dans la table ; sur GCP, regardez `num_undelivered_messages` dans Monitoring.

## La carte du déploiement

En haut de l’interface, l’architecture cible est dessinée bloc par bloc. Chaque bloc s’allume selon ce que l’application constate elle-même, avec la preuve affichée dessous :

| Couleur | Sens |
| --- | --- |
| vert, « prouvé » | l’application l’a vérifié elle-même |
| orange, « à revoir » | ça fonctionne, mais c’est un anti-pattern connu |
| rouge, « en échec » | l’application a essayé et ça ne marche pas |
| pointillés, « pas encore détecté » | rien de visible pour l’instant |
| gris, « à prouver vous-même » | invisible depuis l’application : montrez-le dans la console |
| violet, « simulé en local » | l’équivalent local, en attendant le déploiement |

La carte constate, elle ne note pas : un bloc vert ne dit pas que votre choix est le bon, seulement qu’il est en place.

Pour MoveNow, en mode `bigquery`, le tableau de bord lit :

| Bloc | Comment | Droit nécessaire |
| --- | --- | --- |
| Producteur | positions écrites dans la dernière heure | `roles/bigquery.dataViewer` et `roles/bigquery.jobUser`, déjà nécessaires |
| Table BigQuery | partitionnement et comparaison du schéma au contrat du producteur | idem |
| Topic, BigQuery subscription, dead-letter | configuration de la subscription donnée par `SUBSCRIPTION` | `roles/pubsub.viewer` sur la subscription, facultatif |
| Subscription d’inspection | lecture sans acquittement | `roles/pubsub.subscriber` sur elle |

| Variable | Rôle |
| --- | --- |
| `SUBSCRIPTION` | Facultatif. La subscription de transfert, pour afficher sa configuration sur la carte. |
| `DEMO_PUBLIQUE` | `true` pour les démos hébergées : un transfert coupé se rétablit seul au bout d’une minute. |

flowchart TB
 P["Producteur de positions fictives"] --> T["Topic Pub/Sub"]
 T --> S["Subscription
retry et rétention"]
 S --> W["Transfert BigQuery
natif ou consommateur justifié"]
 W --> BQ["BigQuery
table partitionnée"]
 S -->|"échecs persistants"| DL["Dead-letter topic"]
 DL --> DS["Dead-letter subscription
inspection et reprise"]
 SA["Service agent ou identité
droits explicites"] -.-> S
 SA -.-> BQ
 S -.-> M["Backlog, retard et alertes"]

• L’hypothèse et le résultat attendu.
• Le périmètre, la charge, l’heure de début et l’heure de fin.
• Ce que vous avez observé avant, pendant et après, et l’impact sur le parcours utilisateur.
• Les pistes de diagnostic, les vérifications faites et la cause retenue.
• La correction, la preuve du retour au service et les limites de votre conclusion.

Cette liste est un point de départ et votre architecture demandera d’autres ressources. Chaque lien ouvre la page du provider Google, avec ses arguments et ses exemples.

google_project_service active les API nécessaires, et google_service_account crée une identité dédiée. Donnez à chaque identité les permissions dont elle a besoin, au bon niveau, plutôt qu’un rôle Owner ou Editor sur tout le projet.

Pour l’observabilité, il vous faudra un dashboard, une politique d’alerte et un canal de notification. Vérifiez que la notification arrive réellement.

terraform {
  required_providers {
    google = {
      source  = "hashicorp/google"
      version = "8.6.0"
    }
  }
}

provider "google" {
  project = var.project_id
  region  = var.region
}

Cette version est un exemple, vérifié lors de la préparation du support. Contrôlez la dernière version publiée avant le TP. Déclarez vos variables et commitez .terraform.lock.hcl. Si votre projet utilise déjà une autre version, n’en changez pas sans relire le plan et les notes de migration. En local, utilisez ADC ; en CI, une identité fédérée. Jamais de clé JSON dans le code.

Documentation du provider Google