flowdeck/app/services/realtime_merge.py

"""FlowDeck — v6.4.0 Realtime: résolution de conflits au-delà du last-write-wins.

Le LWW par bloc (v5.13.0) écrase intégralement le bloc du dernier arrivé : si deux
utilisateurs tapent dans le *même* bloc, la saisie du premier est perdue. Ce module
implémente un vrai *merge à trois versions* (diff3-lite) :

    base    = l'état du bloc dont le client dérive sa saisie (envoyé avec l'op)
    current = l'état actuel du bloc côté serveur (déjà mis à jour par d'autres)
    incoming = la nouvelle proposition du client

Règle champ-par-champ :
  * incoming == base   → le client n'a pas touché ce champ      → on garde current
  * current  == base   → le serveur n'a pas touché ce champ     → on garde incoming
  * current  == incoming → les deux ont fait la même chose      → sans conflit
  * sinon (conflit)
      - champ texte (str) : merge de caractères. Les régions modifiées qui ne se
        chevauchent pas sont *toutes conservées* (les deux saisies survivent) ;
        chevauchement réel → LWW sur ce champ + drapeau de conflit.
      - autre type (bool, nombre…) : LWW sur ce champ + drapeau de conflit.

Toutes les fonctions sont pures et testables sans WebSocket ni base de données.
"""
from __future__ import annotations

from typing import Any

__all__ = ["merge_text_3way", "merge_block_3way", "changed_region"]


def changed_region(base: str, other: str) -> tuple[int, int, str] | None:
    """Région de `base` remplacée par `other` (trim préfixe/suffixe commun).

    Retourne ``(start, end, replacement)`` tel que ``base[:start] + replacement +
    base[end:] == other``, ou ``None`` si ``other == base`` (aucun changement).
    """
    if base == other:
        return None
    minlen = min(len(base), len(other))
    prefix = 0
    while prefix < minlen and base[prefix] == other[prefix]:
        prefix += 1
    suffix = 0
    # ne jamais chevaucher le préfixe déjà consommé
    while (suffix < len(base) - prefix and suffix < len(other) - prefix
           and base[len(base) - 1 - suffix] == other[len(other) - 1 - suffix]):
        suffix += 1
    return prefix, len(base) - suffix, other[prefix:len(other) - suffix]


def merge_text_3way(base: str, current: str, incoming: str) -> tuple[str, bool]:
    """Merge à trois versions d'une chaîne. Retourne ``(texte, conflit)``.

    Les éditions qui ne se chevauchent pas sont toutes les deux conservées ;
    un chevauchement réel retombe en LWW (``incoming`` gagne) et signale le conflit.
    """
    if current == incoming:
        return current, False
    if current == base:
        return incoming, False
    if incoming == base:
        return current, False

    rc = changed_region(base, current)
    ri = changed_region(base, incoming)
    if rc is None:
        return incoming, False
    if ri is None:
        return current, False

    c_start, c_end, c_text = rc
    i_start, i_end, i_text = ri

    # Régions disjointes (ou juste adjacentes) → appliquer les deux sur base.
    if c_end <= i_start or i_end <= c_start:
        edits = sorted([(c_start, c_end, c_text), (i_start, i_end, i_text)],
                       key=lambda e: e[0])
        out: list[str] = []
        pos = 0
        for start, end, text in edits:
            if start < pos:
                continue  # sécurité: ne jamais réappliquer par-dessus
            out.append(base[pos:start])
            out.append(text)
            pos = end
        out.append(base[pos:])
        return "".join(out), False

    # Chevauchement réel → LWW sur ce champ, conflit signalé.
    return incoming, True


def _scalar_conflict(base: Any, current: Any, incoming: Any) -> tuple[Any, bool]:
    """Conflit sur un champ non-texte : LWW (incoming gagne)."""
    if current == incoming:
        return current, False
    if current == base:
        return incoming, False
    if incoming == base:
        return current, False
    return incoming, True


def merge_block_3way(base_blk: Any, current_blk: Any,
                     incoming_blk: Any) -> tuple[dict, list[str]]:
    """Merge à trois versions d'un bloc entier.

    Retourne ``(bloc fusionné, champs en conflit)``. Ne lève jamais d'exception :
    une entrée non-dict retombe en LWW (``incoming``) avec conflit signalé sur
    ``__block__`` pour que l'appelant puisse journaliser.
    """
    if not isinstance(base_blk, dict):
        base_blk = {}
    if not isinstance(current_blk, dict):
        current_blk = {}
    if not isinstance(incoming_blk, dict):
        # proposition invalide → on garde l'état serveur
        return dict(current_blk), ["__block__"]

    keys = set(base_blk) | set(current_blk) | set(incoming_blk)
    merged: dict[str, Any] = {}
    conflicts: list[str] = []

    for key in keys:
        b = base_blk.get(key)
        c = current_blk.get(key)
        i = incoming_blk.get(key)

        if i == b:
            # le client n'a pas modifié ce champ → valeur serveur. Si le serveur
            # a *supprimé* le champ (absent de current) alors la suppression doit
            # gagner : on n'insère pas de clé fantôme value=None.
            if key not in current_blk:
                continue
            merged[key] = c
        elif c == b:
            # le serveur n'a pas modifié ce champ → valeur client
            merged[key] = i
        elif c == i:
            merged[key] = c
        else:
            # les deux ont changé, différemment
            if isinstance(b, str) and isinstance(c, str) and isinstance(i, str):
                text, conflict = merge_text_3way(b, c, i)
                merged[key] = text
                if conflict:
                    conflicts.append(key)
            else:
                value, conflict = _scalar_conflict(b, c, i)
                merged[key] = value
                if conflict:
                    conflicts.append(key)

    # Un champ supprimé par le serveur et absent de la proposition client doit
    # rester supprimé (pas de réapparition d'une valeur None fantôme).
    merged = {k: v for k, v in merged.items()
              if not (v is None and k not in incoming_blk)}

    if "id" not in merged:
        # garantir l'identité du bloc même si base était vide
        ident = incoming_blk.get("id") or current_blk.get("id")
        if ident:
            merged["id"] = ident

    return merged, conflicts