# Syncing


<!-- WARNING: THIS FILE WAS AUTOGENERATED! DO NOT EDIT! -->

AnkiWeb’s protocol is simpler than its lack of documentation suggests:
JSON over POST, zstd-compressed, with auth in a header. A sync session
walks a fixed sequence of endpoints under `{endpoint}sync/`: `meta` to
compare collection state, then for a normal (delta) sync `start`,
`applyGraves`, `applyChanges`, `chunk` (repeated), `applyChunk`
(repeated), `sanityCheck2` and `finish`. When the two sides have
diverged too far, a full sync replaces one side’s whole sqlite file via
`upload` or `download`. Everything here was built against Anki’s Rust
implementation as the reference, and is tested below against a real Anki
sync server, with Anki itself as the second client.

``` python
import zstandard, httpx, random, time, json
from fastcore.utils import *
from fastanki.schema import *
from fastanki.collection import *
from fastanki._proto import notetypes_pb2, decks_pb2, deck_config_pb2
```

``` python
import tempfile
from fastcore.test import *
from google.protobuf.json_format import MessageToDict
```

## The wire

Every request is a POST whose body is zstd-compressed JSON, with an
`anki-sync` header carrying the protocol version (`v`, currently 11),
the auth key (`k`, empty only for login), a client version string (`c`),
and a random session id (`s`). Responses come back zstd-compressed too.
A 308 means “use this other host from now on”: AnkiWeb spreads users
across shards this way, so the new endpoint gets saved with the auth
key.

``` python
SYNC_VER = 11
CLIENT_VER = 'anki,26.05 (fastanki),python'

class SyncServer:
    "One AnkiWeb-protocol server (AnkiWeb itself, or self-hosted)"
    def __init__(self, endpoint=None, hkey=''):
        self.endpoint = endpoint or 'https://sync.ankiweb.net/'
        self.hkey,self.skey = hkey,f'{random.getrandbits(32):08x}'
        self.client = httpx.Client(timeout=60)
    def __call__(self, method, obj=None, data=None):
        "POST `obj` (or raw `data`) to sync/`method`, returning the decompressed response body"
        body = zstandard.compress(data if data is not None else json.dumps(obj or {}).encode())
        hdr = json.dumps(dict(v=SYNC_VER, k=self.hkey, c=CLIENT_VER, s=self.skey))
        r = self.client.post(f'{self.endpoint}sync/{method}', content=body, headers={'anki-sync': hdr, 'content-type': 'application/octet-stream'})
        if r.status_code==308:
            self.endpoint = r.headers['location'].rstrip('/') + '/'
            return self(method, obj, data)
        r.raise_for_status()
        return zstandard.ZstdDecompressor().decompressobj().decompress(r.content) if 'anki-original-size' in r.headers else r.content
    def json(self, method, obj=None): return json.loads(self(method, obj))
    def login(self, user, passw):
        "Exchange credentials for a host key, stored on this server object"
        self.hkey = self.json('hostKey', dict(u=user, p=passw))['key']
        return self.hkey
```

## Schema 11 conversions

Here’s a wrinkle: although both sides store the modern schema, the delta
protocol exchanges notetypes, decks and deck configs in the *2012 file
format*, JSON dicts nicknamed “schema 11”, for compatibility with older
clients. So we need mappings between those dicts and our tables. Two
rules matter for round-trip safety: unknown keys accumulate in an
`other` JSON blob within the modern config so nothing is dropped, and a
handful of fields changed units or sign along the way (`autoplay`
becomes `disable_autoplay`, initial ease moves from thousands to a
float, and so on).

``` python
def _other_in(d, reserved):
    "JSON-encode unreserved keys of `d`, Anki's mechanism for preserving unknown fields"
    o = {k:v for k,v in d.items() if k not in reserved}
    return json.dumps(o).encode() if o else b''

def _other_out(b, reserved):
    if not b: return {}
    return {k:v for k,v in json.loads(b).items() if k not in reserved}

_NT_KEYS = {'id','name','type','mod','usn','sortf','did','tmpls','flds','css','latexPre','latexPost','latexsvg','req','originalStockKind','originalId'}
_FLD_KEYS = {'name','ord','sticky','rtl','plainText','font','size','collapsed','description','excludeFromSearch','id','tag','preventDeletion'}
_TMPL_KEYS = {'name','ord','did','afmt','bafmt','qfmt','bqfmt','bfont','bsize','id'}
_REQKIND = {'none':0,'any':1,'all':2}

def nt_from_s11(m):
    "Convert a schema-11 notetype dict to (notetype_row, field_rows, template_rows)"
    c = notetypes_pb2.Notetype.Config(kind=int(m['type']), sort_field_idx=int(m['sortf']), css=m.get('css',''),
        target_deck_id_unused=int(m.get('did') or 0), latex_pre=m.get('latexPre',''), latex_post=m.get('latexPost',''),
        latex_svg=bool(m.get('latexsvg')), original_stock_kind=int(m.get('originalStockKind',0)),
        other=_other_in(m, _NT_KEYS))
    if m.get('originalId') is not None: c.original_id = int(m['originalId'])
    for r in m.get('req') or []:
        q = c.reqs.add(); q.card_ord,q.kind = int(r[0]),_REQKIND[r[1]]; q.field_ords.extend(map(int,r[2]))
    ntid = int(m['id'])
    flds = []
    for i,f in enumerate(sorted(m['flds'], key=lambda f:f.get('ord') or 0)):
        fc = notetypes_pb2.Notetype.Field.Config(sticky=bool(f.get('sticky')), rtl=bool(f.get('rtl')),
            plain_text=bool(f.get('plainText')), font_name=f.get('font','Arial'), font_size=int(f.get('size',20)),
            description=f.get('description',''), collapsed=bool(f.get('collapsed')),
            exclude_from_search=bool(f.get('excludeFromSearch')), other=_other_in(f, _FLD_KEYS))
        if f.get('id') is not None: fc.id = int(f['id'])
        if f.get('tag') is not None: fc.tag = int(f['tag'])
        if f.get('preventDeletion'): fc.prevent_deletion = True
        flds.append((ntid, i, f['name'], fc.SerializeToString()))
    tmpls = []
    for i,t in enumerate(sorted(m['tmpls'], key=lambda t:t.get('ord') or 0)):
        tc = notetypes_pb2.Notetype.Template.Config(q_format=t.get('qfmt',''), a_format=t.get('afmt',''),
            q_format_browser=t.get('bqfmt',''), a_format_browser=t.get('bafmt',''),
            target_deck_id=int(t.get('did') or 0), browser_font_name=t.get('bfont',''),
            browser_font_size=int(t.get('bsize') or 0), other=_other_in(t, _TMPL_KEYS))
        if t.get('id') is not None: tc.id = int(t['id'])
        tmpls.append((ntid, i, t['name'], 0, 0, tc.SerializeToString()))
    row = (ntid, m['name'], int(m['mod']), int(m['usn']), c.SerializeToString())
    return row, flds, tmpls

def nt_to_s11(con, ntid):
    "Convert notetype `ntid`'s rows to a schema-11 dict"
    name,mtime,usn,blob = con.execute('select name, mtime_secs, usn, config from notetypes where id=?',(ntid,)).fetchone()
    c = notetypes_pb2.Notetype.Config(); c.ParseFromString(blob)
    m = dict(id=ntid, name=name, type=c.kind, mod=mtime, usn=usn, sortf=c.sort_field_idx,
        did=c.target_deck_id_unused or None, css=c.css, latexPre=c.latex_pre, latexPost=c.latex_post,
        latexsvg=c.latex_svg, originalStockKind=c.original_stock_kind,
        req=[[r.card_ord, ['none','any','all'][r.kind], list(r.field_ords)] for r in c.reqs],
        flds=[], tmpls=[], **_other_out(c.other, _NT_KEYS))
    if c.HasField('original_id'): m['originalId'] = c.original_id
    for ord_,fname,blob in con.execute('select ord, name, config from fields where ntid=? order by ord',(ntid,)):
        fc = notetypes_pb2.Notetype.Field.Config(); fc.ParseFromString(blob)
        f = dict(name=fname, ord=ord_, sticky=fc.sticky, rtl=fc.rtl, plainText=fc.plain_text,
            font=fc.font_name, size=fc.font_size, description=fc.description, collapsed=fc.collapsed,
            excludeFromSearch=fc.exclude_from_search, **_other_out(fc.other, _FLD_KEYS))
        if fc.HasField('id'): f['id'] = fc.id
        if fc.HasField('tag'): f['tag'] = fc.tag
        f['preventDeletion'] = fc.prevent_deletion
        m['flds'].append(f)
    for ord_,tname,blob in con.execute('select ord, name, config from templates where ntid=? order by ord',(ntid,)):
        tc = notetypes_pb2.Notetype.Template.Config(); tc.ParseFromString(blob)
        t = dict(name=tname, ord=ord_, qfmt=tc.q_format, afmt=tc.a_format, bqfmt=tc.q_format_browser,
            bafmt=tc.a_format_browser, did=tc.target_deck_id or None, bfont=tc.browser_font_name,
            bsize=tc.browser_font_size, **_other_out(tc.other, _TMPL_KEYS))
        if tc.HasField('id'): t['id'] = tc.id
        m['tmpls'].append(t)
    return m
```

``` python
_DECK_KEYS = {'id','mod','name','usn','lrnToday','revToday','newToday','timeToday','collapsed','browserCollapsed',
              'desc','md','dyn','conf','extendNew','extendRev','reviewLimit','newLimit','reviewLimitToday','newLimitToday','desiredRetention'}

def deck_from_s11(d):
    "Convert a schema-11 deck dict to a `decks` table row (filtered decks unsupported: kept as-is if present locally)"
    assert not int(d.get('dyn') or 0), "Filtered decks not supported"
    today = {k: d.get(k) or [0,0] for k in ('lrnToday','revToday','newToday','timeToday')}
    max_day = max(today['timeToday'][0], today['newToday'][0], today['revToday'][0])
    amt = lambda k: today[k][1] if today[k][0]==max_day else 0
    common = decks_pb2.Deck.Common(study_collapsed=bool(d.get('collapsed')), browser_collapsed=bool(d.get('browserCollapsed')),
        last_day_studied=max(max_day,0), new_studied=amt('newToday'), review_studied=amt('revToday'),
        learning_studied=amt('lrnToday'), milliseconds_studied=today['timeToday'][1],
        other=_other_in(d, _DECK_KEYS))
    normal = decks_pb2.Deck.Normal(config_id=int(d.get('conf',1)), extend_new=max(int(d.get('extendNew') or 0),0),
        extend_review=max(int(d.get('extendRev') or 0),0), markdown_description=bool(d.get('md')),
        description=d.get('desc',''))
    for k,f in [('reviewLimit','review_limit'),('newLimit','new_limit')]:
        if d.get(k) is not None: setattr(normal, f, int(d[k]))
    if d.get('desiredRetention') is not None: normal.desired_retention = int(d['desiredRetention'])/100.0
    kind = decks_pb2.Deck.KindContainer(normal=normal)
    return (int(d['id']), d['name'].replace('::','\x1f'), int(d['mod']), int(d['usn']), common.SerializeToString(), kind.SerializeToString())

def deck_to_s11(con, did):
    "Convert deck `did`'s row to a schema-11 dict"
    did,name,mtime,usn,cb,kb = con.execute('select * from decks where id=?',(did,)).fetchone()
    c = decks_pb2.Deck.Common(); c.ParseFromString(cb)
    k = decks_pb2.Deck.KindContainer(); k.ParseFromString(kb)
    assert k.WhichOneof('kind')=='normal', "Filtered decks not supported"
    n = k.normal
    day = c.last_day_studied
    d = dict(id=did, mod=mtime, name=name.replace('\x1f','::'), usn=usn,
        lrnToday=[day,c.learning_studied], revToday=[day,c.review_studied],
        newToday=[day,c.new_studied], timeToday=[day,c.milliseconds_studied],
        collapsed=c.study_collapsed, browserCollapsed=c.browser_collapsed,
        desc=n.description, dyn=0, conf=n.config_id or 1,
        extendNew=n.extend_new, extendRev=n.extend_review, **_other_out(c.other, _DECK_KEYS))
    if n.markdown_description: d['md'] = True
    if n.HasField('review_limit'): d['reviewLimit'] = n.review_limit
    if n.HasField('new_limit'): d['newLimit'] = n.new_limit
    if n.HasField('desired_retention'): d['desiredRetention'] = round(n.desired_retention*100)
    return d
```

``` python
_DCONF_KEYS = {'id','mod','name','usn','maxTaken','autoplay','timer','replayq','new','rev','lapse','dyn',
    'newMix','newPerDayMinimum','interdayLearningMix','reviewOrder','newSortOrder','newGatherPriority',
    'buryInterdayLearning','fsrsWeights','fsrsParams5','fsrsParams6','desiredRetention','ignoreRevlogsBeforeDate',
    'easyDaysPercentages','stopTimerOnAnswer','secondsToShowQuestion','secondsToShowAnswer','questionAction',
    'answerAction','waitForAudio','sm2Retention','weightSearch'}
_NEW_KEYS = {'bury','delays','initialFactor','ints','order','perDay'}
_REV_KEYS = {'bury','ease4','ivlFct','maxIvl','perDay','hardFactor'}
_LAPSE_KEYS = {'delays','leechAction','leechFails','minInt','mult'}

def dconf_from_s11(d):
    "Convert a schema-11 deck config dict to a `deck_config` table row"
    new,rev,lapse = (d.get(k) or {} for k in ('new','rev','lapse'))
    other = {k:v for k,v in d.items() if k not in _DCONF_KEYS}
    for k,sub,keys in [('new',new,_NEW_KEYS),('rev',rev,_REV_KEYS),('lapse',lapse,_LAPSE_KEYS)]:
        o = {kk:vv for kk,vv in sub.items() if kk not in keys}
        if o: other[k] = o
    ints = (new.get('ints') or [1,4])
    c = deck_config_pb2.DeckConfig.Config(
        learn_steps=[float(x) for x in new.get('delays') or [1.0,10.0]],
        relearn_steps=[float(x) for x in lapse.get('delays') or [10.0]],
        new_per_day=int(new.get('perDay',20)), reviews_per_day=int(rev.get('perDay',200)),
        new_per_day_minimum=int(d.get('newPerDayMinimum') or 0),
        initial_ease=int(new.get('initialFactor',2500))/1000.0,
        easy_multiplier=float(rev.get('ease4',1.3)), hard_multiplier=float(rev.get('hardFactor',1.2)),
        lapse_multiplier=float(lapse.get('mult',0.0)), interval_multiplier=float(rev.get('ivlFct',1.0)),
        maximum_review_interval=int(rev.get('maxIvl',36500)), minimum_lapse_interval=int(lapse.get('minInt',1)),
        graduating_interval_good=int(ints[0]), graduating_interval_easy=int(ints[1]),
        new_card_insert_order=int(new.get('order',1)),
        new_card_gather_priority=int(d.get('newGatherPriority') or 0),
        new_card_sort_order=int(d.get('newSortOrder') or 0),
        review_order=int(d.get('reviewOrder') or 0), new_mix=int(d.get('newMix') or 0),
        interday_learning_mix=int(d.get('interdayLearningMix') or 0),
        leech_action=int(lapse.get('leechAction',1)), leech_threshold=int(lapse.get('leechFails',8)),
        disable_autoplay=not d.get('autoplay',True), cap_answer_time_to_secs=max(int(d.get('maxTaken',60)),0),
        show_timer=bool(d.get('timer')), stop_timer_on_answer=bool(d.get('stopTimerOnAnswer')),
        seconds_to_show_question=float(d.get('secondsToShowQuestion') or 0),
        seconds_to_show_answer=float(d.get('secondsToShowAnswer') or 0),
        question_action=int(d.get('questionAction') or 0), answer_action=int(d.get('answerAction') or 0),
        wait_for_audio=d.get('waitForAudio',True), skip_question_when_replaying_answer=not d.get('replayq',True),
        bury_new=bool(new.get('bury')), bury_reviews=bool(rev.get('bury')),
        bury_interday_learning=bool(d.get('buryInterdayLearning')),
        fsrs_params_4=[float(x) for x in d.get('fsrsWeights') or []],
        fsrs_params_5=[float(x) for x in d.get('fsrsParams5') or []],
        fsrs_params_6=[float(x) for x in d.get('fsrsParams6') or []],
        ignore_revlogs_before_date=d.get('ignoreRevlogsBeforeDate',''),
        easy_days_percentages=[float(x) for x in d.get('easyDaysPercentages') or [1.0]*7],
        desired_retention=float(d.get('desiredRetention') or 0.9),
        historical_retention=float(d.get('sm2Retention') or 0.9),
        param_search=d.get('weightSearch',''),
        other=json.dumps(other).encode() if other else b'')
    return (int(d['id']), d['name'], int(d['mod']), int(d['usn']), c.SerializeToString())

def dconf_to_s11(con, dcid):
    "Convert deck config `dcid`'s row to a schema-11 dict"
    dcid,name,mtime,usn,blob = con.execute('select * from deck_config where id=?',(dcid,)).fetchone()
    c = deck_config_pb2.DeckConfig.Config(); c.ParseFromString(blob)
    other = json.loads(c.other) if c.other else {}
    d = dict(id=dcid, mod=mtime, name=name, usn=usn, maxTaken=c.cap_answer_time_to_secs,
        autoplay=not c.disable_autoplay, timer=1 if c.show_timer else 0, replayq=not c.skip_question_when_replaying_answer,
        new=dict(bury=c.bury_new, delays=list(c.learn_steps), initialFactor=round(c.initial_ease*1000),
            ints=[c.graduating_interval_good, c.graduating_interval_easy, 0], order=c.new_card_insert_order,
            perDay=c.new_per_day, **(other.pop('new',{}) if isinstance(other.get('new'),dict) else {})),
        rev=dict(bury=c.bury_reviews, ease4=c.easy_multiplier, ivlFct=c.interval_multiplier,
            maxIvl=c.maximum_review_interval, perDay=c.reviews_per_day, hardFactor=c.hard_multiplier,
            **(other.pop('rev',{}) if isinstance(other.get('rev'),dict) else {})),
        lapse=dict(delays=list(c.relearn_steps), leechAction=c.leech_action, leechFails=c.leech_threshold,
            minInt=c.minimum_lapse_interval, mult=c.lapse_multiplier,
            **(other.pop('lapse',{}) if isinstance(other.get('lapse'),dict) else {})),
        dyn=False, newMix=c.new_mix, newPerDayMinimum=c.new_per_day_minimum,
        interdayLearningMix=c.interday_learning_mix, reviewOrder=c.review_order,
        newSortOrder=c.new_card_sort_order, newGatherPriority=c.new_card_gather_priority,
        buryInterdayLearning=c.bury_interday_learning, fsrsWeights=list(c.fsrs_params_4),
        fsrsParams5=list(c.fsrs_params_5), fsrsParams6=list(c.fsrs_params_6),
        desiredRetention=c.desired_retention, ignoreRevlogsBeforeDate=c.ignore_revlogs_before_date,
        easyDaysPercentages=list(c.easy_days_percentages), stopTimerOnAnswer=c.stop_timer_on_answer,
        secondsToShowQuestion=c.seconds_to_show_question, secondsToShowAnswer=c.seconds_to_show_answer,
        questionAction=c.question_action, answerAction=c.answer_action, waitForAudio=c.wait_for_audio,
        sm2Retention=c.historical_retention, weightSearch=c.param_search)
    d.update(other)
    return d
```

To test the conversions, we get genuine schema-11 dicts from Anki itself
(its Python API still hands out models and decks in exactly this
format), push them through `*_from_s11`, read them back with `*_to_s11`,
and check every value Anki sent is preserved:

``` python
from anki.collection import Collection as AnkiCollection
import anki.lang
```

``` python
def sub_eq(a, b, path=''):
    "Assert every key/value in `a` appears in `b` (recursively, with numeric tolerance)"
    for k,v in a.items():
        w = b.get(k)
        if isinstance(v,dict): sub_eq(v, w or {}, f'{path}{k}.')
        elif isinstance(v,float) or isinstance(w,float): assert abs(v-(w or 0))<1e-6, f'{path}{k}: {v} vs {w}'
        elif isinstance(v,bool) or isinstance(w,bool): assert bool(v)==bool(w), f'{path}{k}: {v} vs {w}'
        else: assert v==w, f'{path}{k}: {v!r} vs {w!r}'

td = Path(tempfile.mkdtemp())
ac = AnkiCollection(str(td/'oracle.anki2'))
mcon = connect(td/'scratch.anki2')
mcon.execute(SCHEMA)
m11 = ac.models.by_name('Cloze')
row,flds,tmpls = nt_from_s11(m11)
mcon.execute('insert into notetypes values (?,?,?,?,?)', row)
mcon.executemany('insert into fields values (?,?,?,?)', flds)
mcon.executemany('insert into templates values (?,?,?,?,?,?)', tmpls)
back = nt_to_s11(mcon, row[0])
sub_eq(dict(m11), back)
```

``` python
d11 = ac.decks.get(1)
mcon.execute('insert into decks values (?,?,?,?,?,?)', deck_from_s11(d11))
sub_eq({k:v for k,v in d11.items() if k!='mid'}, deck_to_s11(mcon, 1))
c11 = ac.decks.get_config(1)
mcon.execute('insert into deck_config values (?,?,?,?,?)', dconf_from_s11(c11))
sub_eq(dict(c11), dconf_to_s11(mcon, 1))
```

## Gathering and applying changes

A delta sync moves four kinds of data. Graves (deletion tombstones) go
first in both directions. Then “unchunked changes”: notetypes, decks,
deck configs and tags, sent whole. Then notes, cards and review log
entries stream in chunks of positional JSON arrays. Rows pending sync
are marked `usn=-1`; as each side sends them it rewrites the mark to the
server’s current usn, so both ends converge on the same sequence number
at `finish`.

``` python
GRAVE_CARD,GRAVE_NOTE,GRAVE_DECK = 0,1,2

@patch
def _pending_graves(self:Collection, new_usn):
    "Pending graves as {cards,notes,decks}, rewriting their usn to `new_usn`"
    g = dict(cards=[], notes=[], decks=[])
    for oid,typ in self.q('select oid, type from graves where usn=-1'):
        g[['cards','notes','decks'][typ]].append(oid)
    self.con.execute('update graves set usn=? where usn=-1', (new_usn,))
    return g

def _chunk_graves(g):
    "Yield grave sub-dicts of at most CHUNK_SIZE oids (cards, then notes, then decks), like Anki's take_chunk"
    items = [(k,o) for k in ('cards','notes','decks') for o in g[k]]
    for i in range(0, len(items), CHUNK_SIZE):
        c = dict(cards=[], notes=[], decks=[])
        for k,o in items[i:i+CHUNK_SIZE]: c[k].append(o)
        yield c

@patch
def _apply_graves(self:Collection, g, new_usn):
    for cid in g.get('cards') or []:
        self.con.execute('delete from cards where id=?',(cid,))
        self.con.execute('insert or replace into graves values (?,0,?)',(cid,new_usn))
    for nid in g.get('notes') or []:
        self.con.execute('delete from notes where id=?',(nid,))
        self.con.execute('insert or replace into graves values (?,1,?)',(nid,new_usn))
    for did in g.get('decks') or []:
        self.con.execute('delete from decks where id=?',(did,))
        self.con.execute('insert or replace into graves values (?,2,?)',(did,new_usn))

@patch
def _local_changes(self:Collection, new_usn, local_newer):
    "Pending unchunked changes in wire format, rewriting local usns to `new_usn`"
    models = [nt_to_s11(self.con, r[0]) for r in self.q('select id from notetypes where usn=-1')]
    decks = [deck_to_s11(self.con, r[0]) for r in self.q('select id from decks where usn=-1')]
    dconf = [dconf_to_s11(self.con, r[0]) for r in self.q('select id from deck_config where usn=-1')]
    tags = [r[0] for r in self.q('select tag from tags where usn=-1')]
    for m in models+decks+dconf: m['usn'] = new_usn
    for t in ['notetypes','decks','deck_config','tags']:
        self.con.execute(f'update {t} set usn=? where usn=-1', (new_usn,))
    changes = dict(models=models, decks=[decks,dconf], tags=tags)
    if local_newer:
        changes['conf'] = {k: json.loads(v) for k,v in self.q('select key, val from config')}
        changes['crt'] = self.q1('select crt from col')
    return changes

@patch
def _apply_changes(self:Collection, ch, new_usn):
    "Apply the server's unchunked changes, skipping any object we hold a newer copy of (Anki's mtime merge)"
    def stale(table, oid, mtime):
        cur = self.q1(f'select mtime_secs from {table} where id=?', oid)
        return cur is not None and cur > mtime
    for m in ch.get('models') or []:
        row,flds,tmpls = nt_from_s11(m)
        if stale('notetypes', row[0], row[2]): continue
        self.con.execute('delete from fields where ntid=?',(row[0],))
        self.con.execute('delete from templates where ntid=?',(row[0],))
        self.con.execute('insert or replace into notetypes values (?,?,?,?,?)', row)
        self.con.executemany('insert into fields values (?,?,?,?)', flds)
        self.con.executemany('insert into templates values (?,?,?,?,?,?)', tmpls)
    decks,dconf = ch.get('decks') or [[],[]]
    for d in decks:
        r = deck_from_s11(d)
        if not stale('decks', r[0], r[2]): self.con.execute('insert or replace into decks values (?,?,?,?,?,?)', r)
    for c in dconf:
        r = dconf_from_s11(c)
        if not stale('deck_config', r[0], r[2]): self.con.execute('insert or replace into deck_config values (?,?,?,?,?)', r)
    for t in ch.get('tags') or []:
        self.con.execute('insert into tags values (?,?,0,NULL) on conflict(tag) do update set usn=excluded.usn', (t, new_usn))
    if ch.get('conf') is not None:
        self.con.execute('delete from config')
        for k,v in ch['conf'].items(): self.con.execute('insert into config values (?,0,0,?)',(k, json.dumps(v).encode()))
    if ch.get('crt') is not None: self.con.execute('update col set crt=?',(ch['crt'],))
```

``` python
CHUNK_SIZE = 250
_NOTE_COLS = 'id, guid, mid, mod, usn, tags, flds, sfld, csum, flags, data'
_CARD_COLS = 'id, nid, did, ord, mod, usn, type, queue, due, ivl, factor, reps, lapses, left, odue, odid, flags, data'
_REVLOG_COLS = 'id, cid, usn, ease, ivl, lastIvl, factor, time, type'

@patch
def _register_note_tags(self:Collection):
    "Ensure every tag on a pending note exists in the tags table, so new tags propagate"
    for (tags,) in self.q('select tags from notes where usn=-1'):
        for t in tags.split():
            self.con.execute('insert or ignore into tags values (?,-1,0,NULL)',(t,))

@patch
def _chunkable_ids(self:Collection):
    return {t: [r[0] for r in self.q(f'select id from {t} where usn=-1')] for t in ('revlog','cards','notes')}

@patch
def _take_chunk(self:Collection, ids, new_usn):
    "Build one wire chunk (positional row arrays) from pending `ids`, rewriting their usn"
    chunk,n = {},0
    for t,cols in [('revlog',_REVLOG_COLS),('cards',_CARD_COLS),('notes',_NOTE_COLS)]:
        take,ids[t] = ids[t][:CHUNK_SIZE-n],ids[t][CHUNK_SIZE-n:]
        n += len(take)
        rows = []
        for oid in take:
            r = list(self.con.execute(f'select {cols} from {t} where id=?',(oid,)).fetchone())
            r[cols.split(', ').index('usn')] = new_usn
            if t=='notes': r[7],r[8] = '',''  # sfld/csum are recomputed by the receiver
            rows.append(r)
        if rows: chunk[t] = rows
        self.con.executemany(f'update {t} set usn=? where id=?', [(new_usn,i) for i in take])
    chunk['done'] = not any(ids.values())
    return chunk

@patch
def _apply_chunk(self:Collection, chunk):
    "Apply a server chunk: newer rows win over local pending ones"
    for r in chunk.get('revlog') or []:
        self.con.execute('insert or ignore into revlog values (?,?,?,?,?,?,?,?,?)', r)
    for r in chunk.get('cards') or []:
        cur = self.con.execute('select usn, mod from cards where id=?',(r[0],)).fetchone()
        if cur and cur[0]==-1 and cur[1]>=r[4]: continue
        self.con.execute('insert or replace into cards values (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)', r)
    for r in chunk.get('notes') or []:
        cur = self.con.execute('select usn, mod from notes where id=?',(r[0],)).fetchone()
        if cur and cur[0]==-1 and cur[1]>=r[3]: continue
        flds = r[6]
        vals = flds.split('\x1f')
        sortf = 0
        blob = self.con.execute('select config from notetypes where id=?',(r[2],)).fetchone()
        if blob:
            cfg = notetypes_pb2.Notetype.Config(); cfg.ParseFromString(blob[0])
            sortf = min(cfg.sort_field_idx, len(vals)-1)
        r = list(r)
        r[7],r[8] = strip_html_media(vals[sortf]), field_csum(vals[0])
        self.con.execute('insert or replace into notes values (?,?,?,?,?,?,?,?,?,?,?)', r)
```

## The sync state machine

`sync_collection` ties it together, following the same sequence as
Anki’s client. `meta` decides between nothing-to-do, a delta sync, and a
full sync; a delta walks graves, changes, chunks down, chunks up, sanity
check, finish. The sanity check compares object counts on both ends; on
a mismatch we bump our schema time so the next sync is a full one, which
is exactly what Anki does.

``` python
class SyncRequired(Exception): pass
class FullSyncRequired(SyncRequired): pass
class SanityCheckFailed(Exception): pass

@patch
def _sanity_counts(self:Collection):
    c = lambda t: self.q1(f'select count() from {t}')
    return [[0,0,0], c('cards'), c('notes'), c('revlog'), c('graves'), c('notetypes'), c('decks'), c('deck_config')]

@patch
def sync_meta(self:Collection):
    mod,scm,usn,crt = self.q('select mod, scm, usn, crt from col')[0]
    empty = not self.q1('select exists(select 1 from cards)')
    return dict(mod=mod, scm=scm, usn=usn, ts=int(time.time()), msg='', cont=True, hostNum=0, empty=empty)

@patch
def sync_collection(self:Collection, srv):
    "Delta-sync with `srv`, returning 'noChanges', 'success', or raising `FullSyncRequired`"
    local = self.sync_meta()
    remote = srv.json('meta', dict(v=SYNC_VER, cv=CLIENT_VER))
    if remote.get('msg'): print(remote['msg'])
    if not remote.get('cont', True): raise Exception(f"Server refused sync: {remote.get('msg')}")
    if remote['mod']==local['mod']: return 'noChanges'
    if remote['scm']!=local['scm']: raise FullSyncRequired(local, remote)
    local_newer = local['mod']>remote['mod']
    server_usn = remote['usn']
    try:
        self.con.execute('BEGIN IMMEDIATE')
        rg = srv.json('start', dict(minUsn=local['usn'], lnewer=local_newer))
        for gchunk in _chunk_graves(self._pending_graves(server_usn)):
            srv.json('applyGraves', dict(chunk=gchunk))
        self._apply_graves(rg, server_usn)
        self._register_note_tags()
        rch = srv.json('applyChanges', dict(changes=self._local_changes(server_usn, local_newer)))
        self._apply_changes(rch, server_usn)
        while True:
            chunk = srv.json('chunk')
            self._apply_chunk(chunk)
            if chunk.get('done'): break
        ids = self._chunkable_ids()
        while True:
            chunk = self._take_chunk(ids, server_usn)
            srv.json('applyChunk', dict(chunk=chunk))
            if chunk['done']: break
        resp = srv.json('sanityCheck2', dict(client=self._sanity_counts()))
        if resp['status']!='ok': raise SanityCheckFailed(str(resp))
        new_mtime = srv.json('finish')
        self.con.execute('update col set mod=?, ls=?, usn=?',(new_mtime, new_mtime, server_usn+1))
        self.con.execute('COMMIT')
        return 'success'
    except Exception as e:
        self.con.execute('ROLLBACK')
        try: srv.json('abort')
        except Exception: pass
        # a sanity mismatch means our object counts diverged from the server's; bump scm to force a full sync next time (Anki's recovery path)
        if isinstance(e, SanityCheckFailed):
            self.con.execute('update col set scm=?',(now_ms(),))
        raise
```

## Full sync and login

A full `download` returns the server’s whole sqlite file; `upload` sends
ours (the server integrity-checks it before accepting, answering with
the literal string `OK`). Auth is a host key from `hostKey` plus the
endpoint any 308 redirect settled on, saved as JSON next to the
collection.

``` python
@patch
def _clear_wal(self:Collection):
    "Checkpoint and remove WAL sidecar files, so `self.path` alone is the whole database"
    self.con.execute('pragma wal_checkpoint(truncate)')

@patch
def full_download(self:Collection, srv):
    "Replace the local collection with the server's copy"
    data = srv('download')
    self._clear_wal()
    self.con.close()
    tmp = self.path.with_suffix('.tmp')
    tmp.write_bytes(data)
    con = connect(tmp)
    assert con.execute('select ver from col').fetchone(), "downloaded file failed to open"
    # keep the downloaded file's usn (the server's); zeroing it would make the next delta needlessly re-fetch the whole collection
    con.execute('update col set ls=mod')
    con.close()
    for ext in ['-wal','-shm']: Path(str(tmp)+ext).unlink(missing_ok=True)
    tmp.replace(self.path)
    for ext in ['-wal','-shm']: Path(str(self.path)+ext).unlink(missing_ok=True)
    self.con = connect(self.path)

@patch
def full_upload(self:Collection, srv):
    "Replace the server's collection with ours"
    self.con.execute('update col set scm=?, mod=?, ls=?, usn=usn+1',(now_ms(), now_ms(), now_ms()))
    # Anki's before_upload: notes/cards/revlog keep their usns (only pending -1 cleared), but decks/notetypes/deck_config/tags all reset to 0; col usn increments
    for t in ['notes','cards','revlog']:
        self.con.execute(f'update {t} set usn=0 where usn=-1')
    for t in ['notetypes','decks','deck_config','tags']:
        self.con.execute(f'update {t} set usn=0 where usn!=0')
    self.con.execute('delete from graves')
    self._clear_wal()
    resp = srv('upload', data=self.path.read_bytes())
    if resp!=b'OK': raise Exception(f"Upload failed: {resp[:200]}")

@patch(as_prop=True)
def _auth_path(self:Collection): return self.path.with_name('auth.json')

@patch
def save_auth(self:Collection, srv): self._auth_path.write_text(json.dumps(dict(hkey=srv.hkey, endpoint=srv.endpoint)))

@patch
def load_auth(self:Collection, endpoint=None):
    "A `SyncServer` from saved auth, or None"
    if not self._auth_path.exists(): return None
    a = json.loads(self._auth_path.read_text())
    return SyncServer(endpoint or a['endpoint'], a['hkey'])

@patch
def sync(self:Collection, user=None, passw=None, endpoint=None, upload=False):
    "Sync with AnkiWeb (or `endpoint`): logs in if needed, handles delta and full syncs"
    srv = self.load_auth(endpoint)
    if srv is None or (user and not srv.hkey):
        if not (user and passw): raise ValueError("No saved auth; pass user and passw")
        srv = SyncServer(endpoint)
        srv.login(user, passw)
    try: res = self.sync_collection(srv)
    except FullSyncRequired as e:
        local,remote = e.args
        if upload:
            if local['empty'] and not remote['empty']:
                raise ValueError("Refusing to replace a non-empty server collection with an empty local one") from None
            self.full_upload(srv)
        else:
            if remote['empty'] and not local['empty']:
                raise ValueError("Server collection is empty but local isn't; pass upload=True to replace the server copy") from None
            self.full_download(srv)
        res = 'full sync'
    self.save_auth(srv)
    return res
```

## Against a real server

The `anki` wheel ships the real sync server (the same code AnkiWeb’s
protocol front end is built from). That makes the test story write
itself: fastanki syncs up to a genuine Anki server on localhost, then
Anki, acting as a second client, syncs down and checks what arrived;
then the same in reverse. The cells below tell that story but are marked
`eval: false`: they need a server process, so `tests/test_sync.py` runs
this exact sequence under pytest, where a fixture launches the server in
a subprocess with a temp data folder and kills it afterwards. Nothing
here talks to the real AnkiWeb.

``` python
import subprocess, sys
```

``` python
PORT = 27183
env = dict(SYNC_BASE=str(td/'server'), SYNC_USER1='tester:s3kret', SYNC_HOST='127.0.0.1', SYNC_PORT=str(PORT))
server = subprocess.Popen([sys.executable,'-m','anki.syncserver'], env=os.environ|env)
EP = f'http://127.0.0.1:{PORT}/'
for _ in range(50):
    try: httpx.get(EP); break
    except httpx.ConnectError: time.sleep(0.1)
```

Our client logs in, adds some content, and full-uploads (a fresh server
and a fresh client have different schema stamps, so the first sync is
always a full one):

``` python
col = Collection.open(td/'ours'/'collection.anki2')
n1 = col.add(Front='syncme', Back='please', tags='wire', deck='Sync::Deep')
ncz = col.add(model='Cloze', Text='{{c1::first}} {{c2::second}}')
test_fail(lambda: col.sync_collection(SyncServer(EP, hkey='bad')))
res = col.sync(user='tester', passw='s3kret', endpoint=EP, upload=True)
test_eq(res, 'full sync')
```

``` python
test_eq(col.sync(), 'noChanges')
```

Now Anki plays second client: log in with its own machinery,
full-download, and inspect what we uploaded.

``` python
(td/'second').mkdir()
oc2 = AnkiCollection(str(td/'second'/'collection.anki2'))
auth = oc2.sync_login('tester', 's3kret', endpoint=EP)
st = oc2.sync_collection(auth, sync_media=False)
oc2.close_for_full_sync()
oc2.full_upload_or_download(auth=auth, server_usn=st.server_media_usn, upload=False)
oc2.reopen(after_full_sync=True)
test_eq(sorted(oc2.find_notes('')), sorted([n1.id, ncz.id]))
test_eq(oc2.find_notes('tag:wire'), [n1.id])
test_eq(oc2.find_notes('deck:Sync::Deep'), [n1.id])
onote = oc2.get_note(ncz.id)
test_eq(onote['Text'], '{{c1::first}} {{c2::second}}')
```

Then it makes changes of its own: a new note, a new deck, an edit and a
deletion, delta-synced up with Anki’s client code:

``` python
theirs = oc2.new_note(oc2.models.by_name('Basic'))
theirs['Front'] = 'from the other side'
oc2.add_note(theirs, oc2.decks.add_normal_deck_with_name('Oracle').id)
on = oc2.get_note(n1.id)
on['Back'] = 'edited remotely'
oc2.update_note(on)
oc2.remove_notes([ncz.id])
st = oc2.sync_collection(auth, sync_media=False)
test_eq(st.required, st.NO_CHANGES)
```

And a delta sync brings all three changes to us, passing the server’s
count check on the way:

``` python
test_eq(col.sync(), 'success')
test_eq(col.find_note_ids(Front='from the other side'), [theirs.id])
test_eq(col.get_note(n1.id)['Back'], 'edited remotely')
test_fail(lambda: col.get_note(ncz.id))
test_eq([d for d in col.decks() if d=='Oracle'], ['Oracle'])
```

Finally the reverse direction once more: our deletion and a new tagged
note, verified by the oracle.

``` python
n2 = col.add(Front='round trip', tags=['back','forth'])
col.remove_notes(n1.id)
test_eq(col.sync(), 'success')
st = oc2.sync_collection(auth, sync_media=False)
oc2.close()
oc3 = AnkiCollection(str(td/'second'/'collection.anki2'))
test_eq(sorted(oc3.find_notes('')), sorted([theirs.id, n2.id]))
test_eq(oc3.find_notes('tag:forth'), [n2.id])
oc3.close()
col.close()
```