Répliquer la BDD adverts en sync continue

CREATE TABLE adverts (
  flx_id           TEXT PRIMARY KEY,
  property_id      TEXT NOT NULL,           -- racine de la chaîne de dédup
  raw_payload      JSONB NOT NULL,
  price            INTEGER,
  last_modified_at TIMESTAMPTZ NOT NULL,
  is_online        BOOLEAN NOT NULL DEFAULT TRUE,
  unpublished_at   TIMESTAMPTZ,
  created_at       TIMESTAMPTZ NOT NULL DEFAULT NOW(),
  updated_at       TIMESTAMPTZ NOT NULL DEFAULT NOW()
);
CREATE INDEX idx_adverts_property ON adverts(property_id);
CREATE INDEX idx_adverts_last_mod ON adverts(last_modified_at);

CREATE TABLE advert_price_history (
  id          BIGSERIAL PRIMARY KEY,
  flx_id      TEXT NOT NULL REFERENCES adverts(flx_id),
  price       INTEGER NOT NULL,
  observed_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
);

# Pseudocode handler
on POST /webhooks/fluximmo:
    if header.x-webhook-key != EXPECTED_KEY: return 401
    enqueue(request.body)            # SQS, RabbitMQ, Redis Streams
    return 200

# Pseudocode worker
handle_payload(body):
    # Branche CREATED — AdvertDto complet
    for entry in body.data.created or []:
        for advert in entry.adverts:
            UPSERT INTO adverts (flx_id, property_id, raw_payload, price, last_modified_at, is_online)
                   VALUES (advert.flxId, advert.propertyFlxId, advert, advert.currentPrice.value,
                           advert.lastModifiedAt, advert.isOnline)
            INSERT INTO advert_price_history (flx_id, price, source) VALUES (advert.flxId, advert.currentPrice.value, 'CREATED')

    # Branche UPDATED — DTO réduit, events dérivés client-side
    for entry in body.data.updated or []:
        for advert in entry.adverts:
            prev = SELECT price, is_online FROM adverts WHERE flx_id = advert.flxId
            new_price  = advert.currentPrice.value
            new_online = advert.isOnline

            if prev is None:
                # advert pas en miroir local : log + ignorer (ou alerter pour reconciliation)
                continue

            UPDATE adverts SET price = new_price, is_online = new_online, updated_at = NOW()
                   WHERE flx_id = advert.flxId

            if new_price != prev.price:
                INSERT INTO advert_price_history (flx_id, price, source) VALUES (advert.flxId, new_price, 'PRICE_DERIVED')
            if new_online and not prev.is_online:
                UPDATE adverts SET unpublished_at = NULL WHERE flx_id = advert.flxId
            if not new_online and prev.is_online:
                UPDATE adverts SET unpublished_at = NOW() WHERE flx_id = advert.flxId

-- Une "Property" = ensemble d'adverts partageant le même property_id
SELECT
  property_id,
  count(*) AS sources_count,
  min(last_modified_at) AS first_seen_at,
  max(last_modified_at) AS last_modified_at,
  bool_or(is_online) AS has_any_online,
  -- prix médian sur les adverts encore en ligne
  percentile_cont(0.5) WITHIN GROUP (ORDER BY price) FILTER (WHERE is_online) AS median_price_online
FROM adverts
WHERE property_id IS NOT NULL
GROUP BY property_id;