"""
MeshBay protocol constants and message type definitions.

MNP (Mesh Node Protocol) — v0.1
MHP (Mesh Bay Hub Protocol) — v0.1

All wire messages are length-prefixed msgpack (4-byte big-endian length header).
Every message carries a "v" field for protocol version.

**`req_id` — the correlation id (added 2026-09-07).** A request may carry one;
the reply to it carries the same value back, and nothing else on the wire does.
It is the caller's own key for its pending request, opaque to the node, and
unique only within one connection.

There was none for a long time, and its absence was not neutral. A reply named
its own type and nothing else, so a caller with more than one request in flight
had to work out which one a message answered from the message itself — and the
replies that name nothing (a bare `ack`, and `{"type": "error"}`, which
webrtc_server.py sends from 240 places while two of them say what they are
about) could only be matched by arrival order. That is a guess, wrong whenever
two replies reorder, and it does not fail quietly: one request is resolved with
another's answer while the request that answer belonged to waits out its own
timeout. Live symptom (2026-09-06): a chat send whose reply went astray left
the composer disabled for thirty seconds, and the Chat tab read as frozen.

Both halves are optional and degrade to what came before: a request without one
is answered without one, and a client that gets no id back falls back to
matching by type. Neither side may treat it as authentication or as a sequence
number — it is a label chosen by the peer, and the only thing it decides is
which local promise a reply belongs to.
"""

import os
from dataclasses import dataclass, field

# The wire versions live in meshbay_common/__init__.py — one source, because a
# second copy here said "0.1" while every message on the wire carried "0.2".
# Nothing imported it, which is the only reason it was harmless.
from meshbay_common import MHP_VERSION, MNP_VERSION  # noqa: F401  (re-export)
from meshbay_common.groupbox import PURPOSE_UPLOAD, seal, unseal
from meshbay_common.webcrypto import (
    chunk_key_aes,
    decrypt_chunk_aes,
    encrypt_chunk_aes,
)

# ── MNP message types ─────────────────────────────────────────────────────────

class MNP:
    HANDSHAKE       = "handshake"
    HANDSHAKE_ACK   = "handshake_ack"
    INDEX_SYNC      = "index_sync"       # full Mesh Group Index
    INDEX_DELTA     = "index_delta"      # incremental update
    # Node -> already-connected members: "the operator's node is scanning
    # right now, N/M bytes done". Never the entries themselves (that is
    # INDEX_SYNC/INDEX_DELTA's job) — just enough for a presence dot to
    # animate. Pushed periodically while scanning, and once more on the
    # transition back to idle, so the indicator is guaranteed to turn off.
    INDEX_PROGRESS  = "index_progress"
    FILE_REQUEST    = "file_req"         # request chunk(s)
    FILE_CHUNK      = "file_chunk"       # encrypted chunk response
    # STREAM_SEGMENT ("stream_seg") was removed in MNP 2.0. It served an
    # MPEG-TS segment as base64 **with no encryption at all** — the one message
    # on the content plane that never was under a GEK-derived key. It predates
    # STREAM_DATA, which does the same job properly (`chunk_ciphertext`, keyed
    # per segment), and its browser caller `fetchStreamSegment` was defined and
    # never once invoked. A live handler on both transports, plaintext media,
    # and no client: removed rather than repaired.
    #
    # Not a Double Ratchet message, and never was — finding C1
    # (`docs/MESHBAY_DESIGN.md` §13.1) rejected exactly that for groups. Since
    # MNP 2.0 it is AES-256-GCM under a
    # per-device subkey of the group's chat epoch key, signed over the
    # ciphertext with the sending device's pinned Ed25519 key. There is no
    # plaintext form on the wire (`chatbox.py`, docs/MESHBAY_DESIGN.md §4.5);
    # `format` distinguishes a *stored* pre-2.0 row, which is still served.
    CHAT_MESSAGE    = "chat_msg"         # one chat message, sealed and signed
    CHAT_ATTACHMENT = "chat_attach"      # attachment metadata
    CHAT_HISTORY    = "chat_hist"       # request message history (newest, or before a cursor)
    CHAT_HISTORY_RESPONSE = "chat_hist_resp"  # history response with messages
    # Link unfurl: the node fetches a URL a member pasted and returns an
    # OpenGraph card. Additive (0.12) — an older node just logs "unknown type"
    # and the client shows the bare link, exactly as before.
    LINK_PREVIEW_REQ  = "link_preview_req"    # client → node: unfurl this URL
    LINK_PREVIEW_RESP = "link_preview_resp"   # node → client: card fields, or ok:false
    # Liveness on an *already open* channel. A peer that goes away without
    # closing leaves a DataChannel that still reads as connected until the next
    # real request hangs, and there was no way to ask. This is not a discovery
    # mechanism: opening a connection in order to ping costs a full ICE/DTLS
    # handshake (measured at 0.6-7 s across two ISPs), so presence in the group
    # list comes from the hub's socket registry instead.
    PING            = "ping"
    PONG            = "pong"
    # GEK_REQUEST / GEK_RESPONSE were removed (NS3, and finding L1): the node must
    # never serve the GEK in plaintext. Members obtain it by unwrapping their own
    # ECIES bundle. The constants lingered after the handlers were deleted, leaving
    # the wire contract looking as though the endpoint still existed.
    # Transfer slots. A download is otherwise invisible to the node -- a series
    # of independent file_req messages, with nothing saying one started or
    # ended -- so there is nothing to count and nothing to cap. The lease is
    # that missing object: `tr` is drawn by the client like `upload_id`, covers
    # a job rather than a file, and dies with the connection.
    #
    # One reply type with a state field, not four: a client that must switch on
    # the message type to discover it is still waiting is a client that will get
    # one branch wrong. Carries no filename and no path -- `tr` is opaque,
    # `bytes` and `chunks` are numbers -- so it stays in clear like
    # INDEX_PROGRESS, for the same stated reason.
    TRANSFER_OPEN    = "transfer_open"     # client -> node: I want a slot
    TRANSFER_CLOSE   = "transfer_close"    # client -> node: I am done with it
    TRANSFER_STATE   = "transfer_state"    # node -> client: granted/queued/closed
    FILE_UPLOAD      = "file_upload"       # client pushes file chunk to node
    FILE_UPLOAD_ACK  = "file_upload_ack"   # node acknowledges chunk receipt
    DIR_CREATE       = "dir_create"        # client → node: make a directory
    DIR_CREATE_ACK   = "dir_create_ack"    # node → client: created
    DIR_DELETE       = "dir_delete"        # client → node: remove an empty directory
    DIR_DELETE_ACK   = "dir_delete_ack"    # node → client: removed
    FILE_DELETE      = "file_delete"       # client requests file deletion
    FILE_DELETE_ACK  = "file_delete_ack"   # node confirms deletion
    STREAM_REQUEST   = "stream_req"        # client requests MSE video stream
    STREAM_INIT      = "stream_init"       # node sends codec info + signals stream start
    STREAM_DATA      = "stream_data"       # node sends encrypted fMP4 segment
    STREAM_END       = "stream_end"        # node signals end of stream
    STREAM_MORE      = "stream_more"       # client → node: room for N more segments
    STREAM_STOP      = "stream_stop"       # client → node: nobody is watching any more
    EPHEMERAL_STREAM = "ephemeral_stream"  # reserved — mobile live push
    HANDSHAKE_CHALLENGE  = "handshake_challenge"   # node → client: GEK proof nonce
    HANDSHAKE_RESPONSE   = "handshake_response"    # client → node: HMAC(GEK, nonce)
    ADMIN_CHALLENGE      = "admin_challenge"       # node → client: Ed25519 sign challenge
    ADMIN_RESPONSE       = "admin_response"        # client → node: Ed25519 signature
    # GEK_BUNDLE_STORE was removed with the invite redesign: the node wraps the GEK
    # itself, for a key the recipient proved possession of, so no member ever hands
    # the node key material (C5b, and the H3 substitution it enabled).
    GEK_BUNDLE_FETCH     = "gek_bundle_fetch"      # client → node: request own wrapped GEK
    GEK_BUNDLE_RESP      = "gek_bundle_resp"       # node → client: wrapped GEK bundle
    KEYPAIR_BUNDLE_STORE = "keypair_bundle_store"  # client → node: store encrypted keypair bundle
    #   optional `bundle_enc_recovery` (MNP 0.14): a second copy wrapped under the
    #   account's recovery key (docs/MESHBAY_DESIGN.md §3.6)
    KEYPAIR_BUNDLE_FETCH = "keypair_bundle_fetch"  # client → node: request own keypair bundle
    KEYPAIR_BUNDLE_RESP  = "keypair_bundle_resp"   # node → client: encrypted keypair bundle
    #   carries `bundle_enc_recovery` too when the node has one stored
    KEYPAIR_BUNDLE_DELETE = "keypair_bundle_delete"  # client → node: withdraw own backup
    # Per-account blobs the node holds and cannot read — playlists (MNP 3.1,
    # docs/playlists.md §8.1). `kind` is "playlists" (the manifest) or
    # "playlist:<uuid>" (one playlist's tracks); `blob_enc` is msgpack `bin`,
    # not base64, because these run to hundreds of kilobytes.
    USER_BLOB_STORE  = "user_blob_store"    # client → node: write one blob
    USER_BLOB_FETCH  = "user_blob_fetch"    # client → node: read one blob
    USER_BLOB_LIST   = "user_blob_list"     # client → node: which kinds, at what rev
    USER_BLOB_DELETE = "user_blob_delete"   # client → node: drop one blob
    USER_BLOB_RESP   = "user_blob_resp"     # node → client: one blob, or null
    USER_BLOB_LIST_RESP = "user_blob_list_resp"  # node → client: kinds and revs only
    JOIN_REQUEST         = "join_request"          # client → node: pair/recognise this identity
    JOIN_RESULT          = "join_result"           # node → client: outcome + wrapped GEK
    INVITE_CREATE        = "invite_create"         # operator → node: issue a pairing code
    INVITE_LINK_CREATE   = "invite_link_create"    # operator → node: a code bound to no account
    INVITE_CANCEL        = "invite_cancel"         # operator → node: take back an unredeemed link
    MEMBER_REVOKE        = "member_revoke"         # operator → node: stop serving the key
    MEMBER_REVOKE_ACK    = "member_revoke_ack"
    MEMBER_UNPIN         = "member_unpin"          # operator → node: forget an identity
    MEMBER_UNPIN_ACK     = "member_unpin_ack"
    APPS_ENABLED         = "apps_enabled"          # operator → node: which group apps to show
    APPS_ENABLED_ACK     = "apps_enabled_ack"
    TRANSFER_LIMITS       = "transfer_limits"       # operator → node: per-member caps here
    TRANSFER_LIMITS_ACK   = "transfer_limits_ack"   # node → this group: the new caps
    SET_SCAN_SETTINGS     = "set_scan_settings"     # operator → node: reconcile/debounce timing
    SET_SCAN_SETTINGS_ACK = "set_scan_settings_ack"
    MEDIA_META_REQ        = "media_meta_req"        # client → node: TMDB metadata for a path
    MEDIA_META_RESP       = "media_meta_resp"       # node → client: TMDB metadata (or none)
    TMDB_CONFIG           = "tmdb_config"            # operator → node: token/language, node-wide
    TMDB_CONFIG_ACK       = "tmdb_config_ack"        # node → everyone: config, never the token
    TMDB_ENABLED          = "tmdb_enabled"           # operator → node: TMDB on/off here
    TMDB_ENABLED_ACK      = "tmdb_enabled_ack"       # node → this group: TMDB on/off here
    SEASON_META_REQ       = "season_meta_req"        # client → node: one season's overview
    SEASON_META_RESP      = "season_meta_resp"       # node → client: season fields, or none
    TMDB_SEARCH_REQ       = "tmdb_search_req"        # client → node: candidates for a query
    TMDB_SEARCH_RESP      = "tmdb_search_resp"       # node → client: id, title, year, poster
    TMDB_OVERRIDE         = "tmdb_override"          # operator → node: replace a match
    TMDB_OVERRIDE_ACK     = "tmdb_override_ack"
    TMDB_REMATCH          = "tmdb_rematch"           # operator → node: drop one file's match
    TMDB_REMATCH_ACK      = "tmdb_rematch_ack"
    # Music app (docs/MESHBAY_DESIGN.md §9.8). Contact is derived from the
    # owner's hub email at login — no config/ack pair needed. Only the
    # per-group toggle remains.
    MUSICBRAINZ_ENABLED = "musicbrainz_enabled"          # operator → node: enable/disable
    MUSICBRAINZ_ENABLED_ACK = "musicbrainz_enabled_ack"  # node → this group: new enabled state
    MUSIC_META_REQ = "music_meta_req"                    # client → node: metadata for a path
    MUSIC_META_RESP = "music_meta_resp"                  # node → client: metadata (or none)
    # A file whose format the browser's own <audio> element cannot decode at
    # all (WMA, Musepack) — resolved to a cached, browser-playable AAC/M4A
    # copy on request, the same "computed once, reused forever" shape as a
    # TMDB poster or a MusicBrainz cover.
    AUDIO_TRANSCODE_REQ = "audio_transcode_req"          # client → node: transcode this file id
    AUDIO_TRANSCODE_RESP = "audio_transcode_resp"        # node → client: cache hash/size/mime
    # One embedded subtitle track, extracted whole-file to WebVTT and cached
    # under its own hash — same shape as the audio transcode above, and for
    # the same reason: the result is small, identical every time, and MSE
    # cannot decode an in-band text track, so it has to travel beside the
    # stream rather than inside it. Whole-file, so a seek re-extracts nothing.
    SUBTITLE_REQ  = "subtitle_req"                       # client → node: file id + track ordinal
    SUBTITLE_RESP = "subtitle_resp"                      # node → client: cache hash/size/mime
    # Device linking. A new device files a request bound to a code it displays;
    # an already-pinned device of the same account approves it. Neither the hub
    # nor the node can produce the countersignature.
    DEVICE_REQUEST       = "device_add_request"    # new device → node
    DEVICE_REQUEST_ACK   = "device_add_request_ack"
    DEVICE_LOOKUP        = "device_lookup"         # approver → node: find by code
    DEVICE_LOOKUP_RESULT = "device_lookup_result"
    DEVICE_ADD           = "device_add"            # approver → node: countersigned
    DEVICE_ADD_ACK       = "device_add_ack"
    DEVICE_LIST          = "device_list"           # anyone → node: my devices
    DEVICE_LIST_RESULT   = "device_list_result"
    DEVICE_REVOKE        = "device_revoke"         # a device retires another
    # "Which of this account's devices am I?" — signed, on an
    # already-authenticated connection. The handshake proves the account and the
    # group; it never proved the device, so the node attributed uploads to the
    # account's oldest key and could not tell one device's chat from another's.
    # Additive (MNP 1.2): a client that stays silent leaves the node exactly
    # where it was.
    DEVICE_HELLO         = "device_hello"          # device → node: this is me
    DEVICE_HELLO_ACK     = "device_hello_ack"
    # Rotation is the half of revocation that revocation cannot do: the node
    # generates a fresh key itself, so no key material crosses the wire.
    GEK_ROTATE           = "gek_rotate"            # operator → node: new group key
    GEK_ROTATE_ACK       = "gek_rotate_ack"
    INVITE_RESULT        = "invite_result"         # node → operator: the code, once
    INVITE_LINK_RESULT   = "invite_link_result"    # node → operator: link code + handle, once
    NODE_STATUS          = "node_status"           # operator → node: list all groups + roots
    NODE_STATUS_ACK      = "node_status_ack"       # node → operator: full status
    ROOT_ADD             = "root_add"              # operator → node: add a directory to a group
    ROOT_ADD_ACK         = "root_add_ack"          # node → operator: confirmed
    ROOT_REMOVE          = "root_remove"           # operator → node: remove a root from a group
    ROOT_REMOVE_ACK      = "root_remove_ack"       # node → operator: confirmed
    # One message for every application's directories, keyed by the app's own
    # name — adding an app adds no message type, and no application has one of
    # its own.
    APP_DIRECTORIES       = "app_directories"      # operator → node: an app's folder(s)
    APP_DIRECTORIES_ACK   = "app_directories_ack"
    # Chat's own two: where attachments are written (a destination, so it must
    # be a read-write root), and whether the node unfurls links members post.
    CHAT_DIRECTORY        = "chat_directory"
    CHAT_DIRECTORY_ACK    = "chat_directory_ack"
    CHAT_LINK_PREVIEW     = "chat_link_preview"
    CHAT_LINK_PREVIEW_ACK = "chat_link_preview_ack"
    # Whether the group's files are listed in members' cross-group Search.
    SEARCH_LISTED         = "search_listed"
    SEARCH_LISTED_ACK     = "search_listed_ack"
    # The keys a group's chat archive is encrypted under, on their way to a
    # member. Sealed under a group-derived subkey, so the payload carries an
    # authentication tag from a key the hub does not hold — and a member who has
    # not completed the handshake is served a ciphertext rather than the keys.
    # Requested rather than pushed on the ack: a group with no chat should not
    # pay for this on every connection.
    CHAT_KEYS_REQ         = "chat_keys_req"
    CHAT_KEYS_RESP        = "chat_keys_resp"
    # Node → this group: a new chat epoch was opened, because somebody was
    # removed. Not a setting — there is no switch; chat is always encrypted
    # (MNP 2.0). Pushed so a connected client stops sealing under the retired
    # key without having to reconnect.
    CHAT_EPOCH            = "chat_epoch"
    CHAT_EPOCH_ACK        = "chat_epoch_ack"
    ROOT_UPDATE          = "root_update"           # operator → node: a root's flags
    ROOT_UPDATE_ACK      = "root_update_ack"
    ROOT_EJECT           = "root_eject"            # operator → node: mark removable root as ejected
    ROOT_EJECT_ACK       = "root_eject_ack"
    ROOT_PLUG            = "root_plug"             # operator → node: re-enable an ejected root
    ROOT_PLUG_ACK        = "root_plug_ack"
    # Member → node: who is in this group and which device keys they hold, with
    # the countersignature that admitted each one. Distinct from ROSTER_READ
    # below, which is the operator's view of the whole node: this is scoped to
    # one group and answers any member of it, because the point is that a member
    # verifies another member's device *for themselves* rather than trusting the
    # node's `sender_id` (Tier 2, docs/MESHBAY_DESIGN.md §3.3).
    GROUP_ROSTER_REQ     = "group_roster_req"
    GROUP_ROSTER_RESP    = "group_roster_resp"
    ROSTER_READ          = "roster_read"            # operator → node: identities + members
    ROSTER_READ_ACK      = "roster_read_ack"
    DENYLIST_READ        = "denylist_read"          # operator → node: show denylist entries
    DENYLIST_READ_ACK    = "denylist_read_ack"
    DENYLIST_CLEAR       = "denylist_clear"         # operator → node: remove denylist entry(ies)
    DENYLIST_CLEAR_ACK   = "denylist_clear_ack"
    GROUP_ATTACH         = "group_attach"           # operator → node: host a new group
    GROUP_ATTACH_ACK     = "group_attach_ack"
    GROUP_DETACH         = "group_detach"           # operator → node: stop hosting a group
    GROUP_DETACH_ACK     = "group_detach_ack"
    NODE_SETTINGS_SET    = "node_settings_set"      # operator → node: change daemon settings
    NODE_SETTINGS_SET_ACK = "node_settings_set_ack"
    NODE_RELOAD          = "node_reload"            # operator → node: re-read node.toml
    NODE_RELOAD_ACK      = "node_reload_ack"


# ── Index entry ───────────────────────────────────────────────────────────────

@dataclass
class IndexEntry:
    id:        str    # blake3 hash of file (hex)
    name:      str    # filename
    path:      str    # path relative to shared directory
    size:      int    # bytes
    type:      str    # video | audio | image | document | archive | other
    added_at:  int    # unix timestamp
    duration:  int | None = None   # seconds, for media
    thumb_hash: str | None = None  # blake3 of thumbnail
    uploader_id: str | None = None  # user_id of who uploaded (None = pre-existing on disk)
    uploader_pk: str | None = None  # Ed25519 public key of uploader (base64 raw 32 bytes)
    width:      int | None = None  # pixels, video/image
    height:     int | None = None  # pixels, video/image
    display_title: str | None = None  # parsed or cleaned-filename title, Videos app
    season:     int | None = None  # parsed season number, Videos app
    episode:    int | None = None  # parsed episode number, Videos app
    artist:     str | None = None  # tag or parsed, Music app
    album:      str | None = None  # tag or parsed, Music app
    track_no:   int | None = None  # tag or parsed, Music app
    taken_at:   int | None = None  # unix timestamp, EXIF DateTimeOriginal — Photos app
    camera:     str | None = None  # "Make Model", when both present — Photos app
    hash_version: int = 1          # 1 = full-file blake3, 2 = partial-read (45 MB sample)


def index_entry_wire(e: IndexEntry) -> dict:
    """
    The wire-dict shape used by INDEX_SYNC/INDEX_DELTA hand-built messages
    (as opposed to GroupIndex.serialize()'s asdict() encoding of the whole
    index). Centralized so the three call sites that build these
    (webrtc_server._do_index_sync, daemon._broadcast_index_change's two
    branches) can't drift from each other as fields are added.
    """
    return {
        "id": e.id, "name": e.name, "path": e.path,
        "size": e.size, "type": e.type, "added_at": e.added_at,
        "uploader_id": e.uploader_id,
        "duration": e.duration, "thumb_hash": e.thumb_hash,
        "width": e.width, "height": e.height,
        "display_title": e.display_title,
        "season": e.season, "episode": e.episode,
        "artist": e.artist, "album": e.album, "track_no": e.track_no,
        "taken_at": e.taken_at, "camera": e.camera,
        "hash_version": e.hash_version,
    }


@dataclass
class IndexDelta:
    base_version: int
    version:      int
    additions:    list[IndexEntry] = field(default_factory=list)
    deletions:    list[str] = field(default_factory=list)  # list of ids
    # Same id as before (same file, same content hash), different field
    # values — e.g. the Videos app's async enrichment filling in duration/
    # thumb_hash/etc. after the file was already indexed with hash+size only.
    # A distinct list from `additions`: `GroupIndex.diff()` only ever adds an
    # id here once it has already appeared unchanged in a prior snapshot.
    updates:      list[IndexEntry] = field(default_factory=list)


# ── File chunk ────────────────────────────────────────────────────────────────
#
# One encoder and one decoder, for every transport.
#
# There used to be two. WebRTC moved to a binary wire format in Phase 9.15 (base64
# costs 33%) and dropped the per-chunk Ed25519 signature with it; the QUIC encoder
# was not brought along, so `file_chunk` meant two different messages depending on
# which transport carried it — base64 fields, hashes and a signature on one, raw
# bytes and a `file_id` on the other. Nothing in the type name said which, and the
# dataclass that used to sit here described only the QUIC half while reading like
# the contract for both. That is finding C6 one size down: two implementations of
# one message, free to drift, with a test for neither.
#
# Why no per-chunk signature: the AES-GCM tag already authenticates the ciphertext
# under a key derived from the GEK, which only group members hold, and since C3 the
# node authenticates itself in the handshake and is pinned by the client. A
# signature per chunk re-proved, once per megabyte, what the session established
# once. (`sign_chunk` still exists in `crypto.py` — it signs the serialized index
# envelope, which is a different artifact; see `indexer/group_index.py`.)


def chunk_ciphertext(
    gek: bytes,
    plaintext: bytes,
    chunk_index: int,
    file_hash: bytes,
) -> tuple[bytes, bytes]:
    """
    `(nonce, ciphertext)` for one chunk.

    Split out because `stream_data` encrypts exactly like `file_chunk` — same key
    derivation, indexed by segment instead of by chunk — and differs only in the
    message it lands in. It used to do so through its own copy of these two lines.
    """
    ckey = chunk_key_aes(gek, file_hash, chunk_index)
    return encrypt_chunk_aes(ckey, plaintext)


def file_chunk_wire(
    gek: bytes,
    plaintext: bytes,
    chunk_index: int,
    file_hash: bytes,
    file_id: str = "",
) -> dict:
    """
    Encrypt one chunk and build the `file_chunk` message.

    `file_hash` is the raw content hash the chunk key is derived from; `file_id` is
    the same value hex-encoded, echoed so a client running several downloads at once
    can tell whose reply arrived. Serving a thumbnail or a cached transcode passes
    the cache blob's own hash for both.
    """
    nonce, ct = chunk_ciphertext(gek, plaintext, chunk_index, file_hash)
    return {
        "type": MNP.FILE_CHUNK,
        "v": MNP_VERSION,
        "file_id": file_id,
        "chunk_index": chunk_index,
        "plaintext_size": len(plaintext),
        "nonce": nonce,
        "ct": ct,
    }


def file_chunk_plaintext(
    gek: bytes,
    msg: dict,
    file_hash: bytes | None = None,
) -> bytes:
    """
    Decrypt a `file_chunk`. Raises `InvalidTag` if the ciphertext was tampered with.

    `file_hash` defaults to the message's own `file_id`, which is what a client that
    asked for a whole file already has. Pass it explicitly only when the caller knows
    better than the peer does.
    """
    if file_hash is None:
        file_hash = bytes.fromhex(msg["file_id"])
    ckey = chunk_key_aes(gek, file_hash, msg["chunk_index"])
    return decrypt_chunk_aes(ckey, msg["nonce"], msg["ct"])


# ── Uploads (MNP 2.0) ─────────────────────────────────────────────────────────
#
# The write path, sealed under the group key the way the read path always was.
# One encoder for both directions, here rather than in the client, for the reason
# `file_chunk` has one: two copies of a wire shape with a single consumer each is
# how `index_sync` and `file_chunk` forked (finding C6), and nothing noticed
# until someone went looking.
#
# What stays in clear, and why each has to:
#   `type`, `v`     — routed and version-checked before anything can be decrypted
#   `upload_id`     — the correlation key. It replaces `filename`, which used to
#                     play that role and cannot any more: naming the file in clear
#                     to match an ack against a request would give back exactly
#                     what the seal is for. Client-chosen, opaque to the node,
#                     unique within one connection; never an authorization input.
#   `chunk_index`   — ordering, which the node enforces before it opens anything
#   `total_chunks`  — how many to expect
#
# `group_id` is *not* on the message: the session already decided which group it
# is on, and the node uses that as the AAD. A client naming its own group here
# would be choosing which key its bytes are checked against.

UPLOAD_ID_LEN = 16   # 128 bits of client-chosen correlation, hex on the wire


def new_upload_id() -> str:
    """A fresh correlation id for one upload."""
    return os.urandom(UPLOAD_ID_LEN).hex()


def file_upload_wire(
    gek: bytes,
    group_id: str,
    *,
    upload_id: str,
    chunk_index: int,
    total_chunks: int,
    filename: str,
    data: bytes,
    dir: str = "",
    root: str = "",
) -> dict:
    """
    One sealed `file_upload` chunk.

    `filename`, `dir` and `root` ride inside the seal with the bytes: sealing the
    content and announcing the name beside it would be theatre. They are repeated
    on every chunk rather than sent once — a hundred bytes against a 48 KiB chunk
    — because a header that arrives once is state the node has to carry, and
    upload state that can disagree with the chunk in hand is what `_free_name` and
    the chunk-ordering rule exist to prevent.
    """
    payload = {"filename": filename, "data": data, "dir": dir, "root": root}
    return {
        "type": MNP.FILE_UPLOAD,
        "v": MNP_VERSION,
        "upload_id": upload_id,
        "chunk_index": chunk_index,
        "total_chunks": total_chunks,
        **seal(gek, PURPOSE_UPLOAD, MNP.FILE_UPLOAD, group_id, payload),
    }


def file_upload_payload(gek: bytes, group_id: str, msg: dict) -> dict:
    """
    Open a `file_upload`. Raises on anything that does not open.

    Never a partial result and never a default: a chunk that does not open is not
    an empty file with an empty name, it is a peer we cannot talk to. `unseal`
    says why at length.
    """
    return unseal(gek, PURPOSE_UPLOAD, MNP.FILE_UPLOAD, group_id, msg)


# "Where am I?", asked as an ordinary sealed upload chunk rather than as a new
# message.
#
# The node identifies an upload by (member, directory, filename), so a client
# resuming one has to name the file — and `transfer_open`, the obvious place to
# ask, travels in clear. Naming it there would undo exactly what sealing the
# upload path bought: before MNP 2.0 the same file was ciphertext leaving a node
# and plaintext arriving at one.
#
# So the question is asked inside the seal that already exists, as a chunk with
# no bytes and this index. The node writes nothing, changes nothing, and answers
# with `resume_from`. A node that predates this refuses the index, which the
# client reads as "start from the beginning" — the behaviour it had anyway.
UPLOAD_PROBE_INDEX = -1


def file_upload_ack_wire(
    gek: bytes,
    group_id: str,
    *,
    upload_id: str,
    chunk_index: int,
    filename: str,
    stored_as: str,
    dir: str = "",
    resume_from: int | None = None,
) -> dict:
    """
    The node's answer to one chunk, sealed the same way.

    `stored_as` is the name the node settled on — it finds a free one rather than
    replacing anything — and `dir` is where it landed. Both name the operator's
    content, so both belong inside the seal; only `upload_id` and `chunk_index`
    stay out, because the client matches on them.

    `resume_from` answers the probe chunk (`UPLOAD_PROBE_INDEX`): how many
    chunks of this file the node already holds. Inside the seal like the rest —
    it is a fact about the operator's disk — and absent from an ordinary ack, so
    a client can tell the two apart without looking at `chunk_index`.
    """
    payload = {"filename": filename, "stored_as": stored_as, "dir": dir}
    if resume_from is not None:
        payload["resume_from"] = int(resume_from)
    return {
        "type": MNP.FILE_UPLOAD_ACK,
        "v": MNP_VERSION,
        "upload_id": upload_id,
        "chunk_index": chunk_index,
        **seal(gek, PURPOSE_UPLOAD, MNP.FILE_UPLOAD_ACK, group_id, payload),
    }


def file_upload_ack_payload(gek: bytes, group_id: str, msg: dict) -> dict:
    """Open a `file_upload_ack`. Raises on anything that does not open."""
    return unseal(gek, PURPOSE_UPLOAD, MNP.FILE_UPLOAD_ACK, group_id, msg)