#!/usr/bin/env python3
"""Schema evolution: old code reads new data, new code reads old. DDIA 2e Ch. 5.

Add a field to a Protobuf/Avro schema and watch compatibility hold -- or break.

  * FORWARD compat (old reads new): a v1 reader decodes v2-written bytes and
    silently ignores the field it doesn't know about.
  * BACKWARD compat (new reads old): a v2 reader decodes v1-written bytes and
    fills the missing field with its default.
  * BUT: adding a field WITHOUT a default breaks backward compatibility -- an
    Avro reader whose schema has a no-default field cannot resolve old bytes
    that lack it, and raises.

Protobuf gives forward+backward compatibility for free (every field is optional
in proto3, unknown tags are skipped, absent fields take a type default). Avro
makes the rule explicit: you may add or remove only a field that has a default.

Run it:
    uv run --python 3.12 --with grpcio-tools --with protobuf --with fastavro \
        python3 schema_evolution.py
(The .proto schemas are compiled at runtime via grpc_tools -- no separate protoc
step. Deterministic -> the transcript reproduces exactly.)
"""
import io
import os
import shutil
import sys
import tempfile


def protobuf_demo(work):
    """Two schema versions of Person; v2 adds an `email` field (tag 3)."""
    v1 = os.path.join(work, "person_v1.proto")
    v2 = os.path.join(work, "person_v2.proto")
    with open(v1, "w") as f:
        f.write('syntax = "proto3";\n'
                'package v1;\n'
                'message Person {\n'
                '  string user_name = 1;\n'
                '  int64  id = 2;\n'
                '}\n')
    with open(v2, "w") as f:
        f.write('syntax = "proto3";\n'
                'package v2;\n'
                'message Person {\n'
                '  string user_name = 1;\n'
                '  int64  id = 2;\n'
                '  string email = 3;\n'
                '}\n')
    from grpc_tools import protoc
    protoc.main(["protoc", f"-I{work}", f"--python_out={work}", v1])
    protoc.main(["protoc", f"-I{work}", f"--python_out={work}", v2])
    sys.path.insert(0, work)
    import person_v1_pb2
    import person_v2_pb2

    print("=== PROTOBUF ===")
    print("v1: message Person { string user_name = 1; int64 id = 2; }")
    print("v2: message Person { string user_name = 1; int64 id = 2; string email = 3; }")
    print()

    # (a) FORWARD compat: encode with v2 (email set) -> decode with v1.
    new = person_v2_pb2.Person(user_name="Martin", id=1337,
                               email="martin@example.com")
    new_bytes = new.SerializeToString()
    old_reader = person_v1_pb2.Person()
    old_reader.ParseFromString(new_bytes)   # succeeds; tag 3 is unknown -> skipped
    print("[old-reads-new] v1 reader on v2 bytes (email set):  OK")
    print(f"    recovered: user_name={old_reader.user_name!r}, id={old_reader.id}")
    print(f"    (email had tag 3, unknown to v1 -> silently ignored; v1 has no email field)")
    print()

    # (b) BACKWARD compat: encode with v1 (no email) -> decode with v2.
    old = person_v1_pb2.Person(user_name="Martin", id=1337)
    old_bytes = old.SerializeToString()
    new_reader = person_v2_pb2.Person()
    new_reader.ParseFromString(old_bytes)   # succeeds; email absent -> "" default
    print("[new-reads-old] v2 reader on v1 bytes (no email):   OK")
    print(f"    recovered: user_name={new_reader.user_name!r}, id={new_reader.id}, "
          f"email={new_reader.email!r}")
    print(f"    (email absent in old bytes -> string type default \"\" filled in)")
    print()


def avro_demo():
    """Writer/reader schema resolution; a field WITH vs WITHOUT a default."""
    import fastavro

    writer_schema = {
        "type": "record", "name": "Person", "fields": [
            {"name": "userName", "type": "string"},
            {"name": "id", "type": "long"},
        ]}
    # New reader schema #1: adds favoriteColor WITH a default -> resolvable.
    reader_with_default = {
        "type": "record", "name": "Person", "fields": [
            {"name": "userName", "type": "string"},
            {"name": "id", "type": "long"},
            {"name": "favoriteColor", "type": "string", "default": "blue"},
        ]}
    # New reader schema #2: adds favoriteColor WITHOUT a default -> breaks.
    reader_no_default = {
        "type": "record", "name": "Person", "fields": [
            {"name": "userName", "type": "string"},
            {"name": "id", "type": "long"},
            {"name": "favoriteColor", "type": "string"},
        ]}

    ws = fastavro.parse_schema(writer_schema)
    record = {"userName": "Martin", "id": 1337}
    buf = io.BytesIO()
    fastavro.schemaless_writer(buf, ws, record)
    old_bytes = buf.getvalue()

    print("=== AVRO ===")
    print("writer schema : { userName: string, id: long }")
    print('reader schema : adds favoriteColor  (once WITH default "blue", once WITHOUT)')
    print()

    # (a) New reader with a DEFAULTED added field -> reads old bytes, fills default.
    rs_ok = fastavro.parse_schema(reader_with_default)
    got = fastavro.schemaless_reader(io.BytesIO(old_bytes), ws, rs_ok)
    print("[new-reads-old, field HAS default] reader adds favoriteColor=\"blue\":  OK")
    print(f"    recovered: {got}")
    print(f"    (old bytes lack favoriteColor -> reader fills the schema default)")
    print()

    # (b) New reader with a NON-defaulted added field -> cannot resolve old bytes.
    rs_bad = fastavro.parse_schema(reader_no_default)
    print("[new-reads-old, field has NO default] reader adds favoriteColor (required):")
    try:
        fastavro.schemaless_reader(io.BytesIO(old_bytes), ws, rs_bad)
        print("    OK  <- unexpected!")
    except Exception as e:
        print(f"    FAILS -> {type(e).__name__}: {e}")
        print(f"    (no default + absent in old bytes -> reader cannot supply the value)")
    print()


def main():
    work = tempfile.mkdtemp(prefix="ddia_schema_evo_")
    try:
        protobuf_demo(work)
        avro_demo()
        print("Rule: you may add/remove a field only if it has a default.")
        print("Protobuf bakes that in (proto3 fields are optional, defaults are typed);")
        print("Avro makes it a hard requirement -- a no-default add breaks old readers.")
    finally:
        shutil.rmtree(work, ignore_errors=True)


if __name__ == "__main__":
    main()