/* -*- Mode: C; tab-width: 4; indent-tabs-mode: nil; c-basic-offset: 2 -*- */ 
/* vim: set ts=2 et sw=2 tw=80: */
/*
** Copyright (c) 2013 D. Richard Hipp
**
** This program is free software; you can redistribute it and/or
** modify it under the terms of the Simplified BSD License (also
** known as the "2-Clause License" or "FreeBSD License".)
**
** This program is distributed in the hope that it will be useful,
** but without any warranty; without even the implied warranty of
** merchantability or fitness for a particular purpose.
**
** Author contact information:
**   drh@hwaci.com
**   http://www.hwaci.com/drh/
**
*******************************************************************************
** This file houses the code for the fsl_content_xxx() APIS.
*/
#include <assert.h>

#include "fossil-scm/fossil-internal.h"

/* Only for debugging */
#include <stdio.h>
#define MARKER(pfexp)                                               \
  do{ printf("MARKER: %s:%d:%s():\t",__FILE__,__LINE__,__func__);   \
    printf pfexp;                                                   \
  } while(0)


fsl_int_t fsl_content_size( fsl_cx * f, fsl_id_t blobRid ){
  fsl_db * dbR = f ? fsl_cx_db_repo(f) : NULL;
  if(!f) return -3;
  else if(blobRid<=0) return -4;
  else if(!dbR) return -5;
  else{
    int rc;
    fsl_int_t rv = -2;
    fsl_stmt * q = NULL;
    rc = fsl_db_prepare_cached(dbR, &q,
                                   "SELECT size FROM blob "
                                   "WHERE rid=?");
    if(!rc){
      rc = fsl_stmt_bind_id(q, 1, blobRid);
      if(!rc){
        if(FSL_RC_STEP_ROW==fsl_stmt_step(q)){
          rv = (fsl_int_t)fsl_stmt_g_int64(q, 0);
        }
      }
      fsl_stmt_cached_yield(q);
    }
    return rv;
  }
}


int fsl_content_blob( fsl_cx * f, fsl_id_t blobRid, fsl_buffer * tgt ){
  fsl_db * dbR = f ? fsl_cx_db_repo(f) : NULL;
  if(!f || !tgt) return FSL_RC_MISUSE;
  else if(blobRid<=0) return FSL_RC_RANGE;
  else if(!dbR) return FSL_RC_NOT_A_REPO;
  else{
    int rc;
    fsl_stmt * q = NULL;
    rc = fsl_db_prepare_cached( dbR, &q,
                                "SELECT content FROM blob "
                                "WHERE rid=? AND size>=0");
    if(!rc){
      rc = fsl_stmt_bind_id(q, 1, blobRid);
      if(!rc && (FSL_RC_STEP_ROW==(rc=fsl_stmt_step(q)))){
        void const * mem = NULL;
        fsl_size_t memLen = 0;
        tgt->used = 0;
        fsl_stmt_get_blob(q, 0, &mem, &memLen);
        if(mem && memLen){
          rc = fsl_buffer_append(tgt, mem, memLen);
          if(!rc && fsl_buffer_is_compressed(tgt)){
            rc = fsl_buffer_uncompress(tgt, tgt);
          }
        }
      }else if(FSL_RC_STEP_DONE==rc){
        rc = FSL_RC_NOT_FOUND;
      }
      fsl_stmt_cached_yield(q);
    }
    if(rc && !f->error.code && dbR->error.code){
      fsl_cx_uplift_db_error(f, dbR);
    }
    return rc;
  }
}


char fsl_content_is_private(fsl_cx * f, fsl_id_t rid){
  fsl_stmt * s1 = NULL;
  fsl_db * db = fsl_cx_db_repo(f);
  int rc = db
    ? fsl_db_prepare_cached(db, &s1,
                            "SELECT 1 FROM private "
                            "WHERE rid=?")
    : FSL_RC_MISUSE;
  if(!rc){
    rc = fsl_stmt_bind_id(s1, 1, rid);
    if(!rc) rc = fsl_stmt_step(s1);
    fsl_stmt_cached_yield(s1);
  }
  return rc==FSL_RC_STEP_ROW;
}


int fsl_content_get( fsl_cx * f, fsl_id_t rid, fsl_buffer * tgt ){
  fsl_db * db = f ? fsl_cx_db_repo(f) : NULL;
  if(!f || !tgt) return FSL_RC_MISUSE;
  else if(rid<=0){
    return fsl_cx_err_set(f, FSL_RC_RANGE,
                          "RID %"FSL_ID_T_PFMT" is out of range.",
                          (fsl_id_t)rid);
  }
  else if(!db){
    return fsl_cx_err_set(f, FSL_RC_NOT_A_REPO,
                          "Fossil has no repo opened.");
  }
  else{
    int rc;
    char gotIt = 0;
    fsl_id_t nextRid;
    fsl_acache * ac = &f->cache.arty;
    tgt->used = 0;
    if(fsl_id_bag_contains(&f->cache.arty.missing, rid)){
      /* Early out if we know the content is not available */
      return FSL_RC_NOT_FOUND;
    }

    /* Look for the artifact in the cache first */
    if( fsl_id_bag_contains(&f->cache.arty.inCache, rid) ){
      fsl_size_t i;
      fsl_acache_line * line;
      for(i=0; i<ac->used; i++){
        line = &ac->list[i];
        if( line->rid==rid ){
          rc = fsl_buffer_copy(&line->content, tgt);
          line->age = ac->nextAge++;
          return rc;
        }
      }
    }

    nextRid = 0;
    rc = fsl_delta_src_id(f, rid, &nextRid);
    /* MARKER(("rc=%d, nextRid=%"FSL_ID_T_PFMT"\n", rc, nextRid)); */
    if(rc) return rc;
    if( nextRid == 0 ){
      /* This is not a delta, so get its raw content. */
      rc = fsl_content_blob(f, rid, tgt);
      gotIt = 0==rc;
    }else{
      /* Looks like a delta, so let's expand it... */
      fsl_int_t n           /* number of used entries in 'a' */;
      fsl_int_t nAlloc = 10 /* number it items allocated in 'a' */;
      fsl_id_t * a = NULL    /* array of rids we expand */;
      fsl_int_t mx;
      fsl_buffer delta = fsl_buffer_empty;
      fsl_buffer next = fsl_buffer_empty  /* delta-applied content */ ;
      assert(nextRid>0);
      a = fsl_malloc( sizeof(a[0]) * nAlloc );
      if(!a) return FSL_RC_OOM;
      a[0] = rid;
      a[1] = nextRid;
      n = 1;
      while( !fsl_id_bag_contains(&ac->inCache, nextRid)
             && !fsl_delta_src_id(f, nextRid, &nextRid)
             && (nextRid>0)){
        /* Figure out how big n needs to be... */
        ++n;
        if( n >= nAlloc ){
          /* Expand 'a' */
          void * remem;
          if( n > fsl_db_g_size(db, 0,
                                "SELECT max(rid) FROM blob")){
            rc = fsl_cx_err_set(f, FSL_RC_RANGE,
                                "Infinite loop in delta table.");
            goto end_delta;
          }
          nAlloc = nAlloc * 2;
          remem = fsl_realloc(a, nAlloc*sizeof(a[0]));
          if(!remem){
            rc = FSL_RC_OOM;
            goto end_delta;
          }
          a = (fsl_id_t*)remem;
        }
        a[n] = nextRid;
      }
      /**
         Recursively expand deltas to get the content...
      */
      mx = n;
      rc = fsl_content_get( f, a[n], tgt );
      /* MARKER(("Getting content for rid #%"FSL_ID_T_PFMT", rc=%d\n", a[n], rc)); */
      --n;
      for( ; !rc && (n>=0); --n){
        rc = fsl_content_blob(f, a[n], &delta);
        /* MARKER(("Getting/applying delta rid #%"FSL_ID_T_PFMT", rc=%d\n", a[n], rc)); */
        if(rc) goto end_delta;
        if(!delta.used){
          assert(!"Is this possible? The v1 tree has a similar "
                 "condition but i naively don't believe it's necessary.");
          continue;
        }
        next = fsl_buffer_empty;
        rc = fsl_buffer_delta_apply2(tgt, &delta, &next, &f->error);
        if(rc) goto end_delta;
#if 0
        /*
           In my (very simple) tests this cache costs us more than it
           saves. TODO: re-test this once we can do a 'rebuild', or
           something more intensive than processing a single
           manifest's R-card. At that point we can set a f->flags bit
           to enable or disable this block for per-use-case
           optimization purposes.
        */
           
        if( (mx-n)%8==0 ){
          rc = fsl_acache_insert( ac, a[n+1], tgt );
          if(rc){
            fsl_buffer_clear(&next);
            goto end_delta;
          }
          assert(!tgt->mem && "Passed to artifact cache.");
        }else{
          fsl_buffer_clear(tgt);
        }
#else
        if(mx){/*unused var*/}
        fsl_buffer_clear(tgt);
#endif
        *tgt = next;
      }
      end_delta:
      fsl_buffer_clear(&delta);
      fsl_free(a);
      gotIt = 0==rc;
    }

    if(!rc){
      /* i'm not 100% sure this bit has been ported
         accurately... */
      rc = fsl_id_bag_insert(gotIt
                             ? &f->cache.arty.available
                             : &f->cache.arty.missing,
                             rid);
    }
    return rc;
  }
}

int fsl_content_get_by_sym( fsl_cx * f, char const * sym, fsl_buffer * tgt ){
  int rc;
  fsl_db * db = f ? fsl_needs_repo(f) : NULL;
  fsl_id_t rid = 0;
  if(!f || !sym || !tgt) return FSL_RC_MISUSE;
  else if(!db) return FSL_RC_NOT_A_REPO;
  rc = fsl_sym_to_rid(f, sym, FSL_ATYPE_ANY, &rid);
  return rc ? rc : fsl_content_get(f, rid, tgt);
}

/**
 ** Mark artifact rid as being available now. Update f's cache to show
 ** that everything that was formerly unavailable because rid was
 ** missing is now available. Returns 0 on success. f must have
 ** an opened repo and rid must be valid.
 */
static int fsl_content_mark_available(fsl_cx * f, fsl_id_t rid){
  fsl_id_bag pending = fsl_id_bag_empty;
  int rc;
  fsl_stmt * st = NULL;
  fsl_db * db = fsl_cx_db_repo(f);
  assert(f);
  assert(db);
  assert(rid>0);
  if( fsl_id_bag_contains(&f->cache.arty.available, rid) ) return 0;
  rc = fsl_id_bag_insert(&pending, rid);
  if(rc) goto end;
  while( (rid = fsl_id_bag_first(&pending))!=0 ){
    fsl_id_bag_remove(&pending, rid);
    rc = fsl_id_bag_insert(&f->cache.arty.available, rid);
    if(rc) goto end;
    fsl_id_bag_remove(&f->cache.arty.missing, rid);
    if(!st){
      rc = fsl_db_prepare_cached(db, &st,
                                 "SELECT rid FROM delta "
                                 "WHERE srcid=?");
      if(rc) goto end;
    }
    rc = fsl_stmt_bind_id(st, 1, rid);
    while( !rc && (FSL_RC_STEP_ROW==fsl_stmt_step(st)) ){
      fsl_id_t const nx = fsl_stmt_g_id(st,0);
      assert(nx>0);
      rc = fsl_id_bag_insert(&pending, nx);
    }
    fsl_stmt_cached_yield(st);
  }
  end:
  if(st) fsl_stmt_cached_yield(st);
  fsl_id_bag_clear(&pending);
  return rc;
}


int fsl_content_put_ex( fsl_cx * f, fsl_buffer const * pBlob,
                        fsl_uuid_cstr zUuid,
                        fsl_id_t srcId,
                        fsl_size_t uncompSize,
                        char isPrivate,
                        fsl_id_t * outRid){
  fsl_size_t size;
  fsl_id_t rid;
  fsl_stmt s1 = fsl_stmt_empty;
  fsl_buffer cmpr = fsl_buffer_empty;
  fsl_buffer hash = fsl_buffer_empty;
  char markAsUnclustered = 0;
  char markAsUnsent = 1;
  char isDephantomize = 0;
  fsl_db * dbR = fsl_cx_db_repo(f);
  int rc;
  char inTrans = 0;
  assert(f);
  assert(dbR);
  assert(pBlob);
  assert(srcId==0 || zUuid!=NULL);
  if(!dbR) return FSL_RC_NOT_A_REPO; 
  if(!zUuid){
    assert(0==uncompSize);
    rc = fsl_sha1sum_buffer(pBlob, &hash);
  }else{
    rc = fsl_buffer_append(&hash, zUuid, -1);
  }
  if(rc) return rc;
  if(uncompSize){
    /* pBlob is assumed to be compressed. */
    assert(fsl_buffer_is_compressed(pBlob));
    size = uncompSize;
  }else{
    size = pBlob->used;
    if(srcId>0){
      rc = fsl_delta_applied_size(pBlob->mem, pBlob->used, &size);
      if(rc) goto end;
    }
  }
  rc = fsl_db_transaction_begin(dbR);
  if(rc) goto end;
  inTrans = 1;

  /* Check to see if the entry already exists and if it does whether
  ** or not the entry is a phantom.
  */
  rc = fsl_db_prepare(dbR, &s1,
                        "SELECT rid, size FROM blob WHERE uuid=%B",
                        &hash);
  if(rc) goto end;
  rc = fsl_stmt_step(&s1);
  switch(rc){
    case FSL_RC_STEP_ROW:
      rid = fsl_stmt_g_id(&s1, 0);
      if( fsl_stmt_g_int64(&s1, 1)>=0 || pBlob==0 ){
        /* Either the entry is not a phantom or it is a phantom but we
        ** have no data with which to dephantomize it.  In either case,
        ** there is nothing for us to do other than return the RID.
        */
        /*
          Reminder: the do-nothing-for-empty-phantom behaviour is
          arguable (but historical). There is a corner case there
          involving an empty file. So far, so good, though.
        */
        fsl_stmt_finalize(&s1);
        fsl_db_transaction_end(dbR,0);
        if(outRid) *outRid = rid;
        return 0;
      }
      break;
    case FSL_RC_STEP_DONE:
      /* No entry with the same UUID currently exists */
      rid = 0;
      markAsUnclustered = 1;
      rc = 0;
      break;
    default:
      assert(0!=rc);
  }
  fsl_stmt_finalize(&s1);
  if(rc) goto end;

#if 0
  /* Requires app-level data. We might need a client hook mechanism or
     other metadata here.
  */
  /* Construct a received-from ID if we do not already have one */
  if( f->cache.rcvid <= 0 ){
    rc = fsl_db_exec(dbR, 
       "INSERT INTO rcvfrom(uid, mtime, nonce, ipaddr)"
       "VALUES(%d, julianday('now'), %Q, %Q)",
       g.userUid, g.zNonce, g.zIpAddr
    );
    f->cache.rcvid = fsl_db_last_insert_id(dbR);
  }
#endif

  if( uncompSize ){
    cmpr = *pBlob;
  }else{
    rc = fsl_buffer_compress(pBlob, &cmpr);
    if(rc) goto end;
  }

  if( rid>0 ){
#if 1
    assert(!"NYI: adding data to phantom. Requires some missing pieces.");
    rc = FSL_RC_NYI;
    goto end;
#else
    /* We are just adding data to a phantom */
    rc = fsl_db_prepare(dbR, &s1,
                        "UPDATE blob SET rcvid=%"FSL_ID_T_PFMT
                        ", size=%"FSL_SIZE_T_PFMT
                        ", content=? WHERE "
                        "rid=%"FSL_ID_T_PFMT,
                        (fsl_id_t)f->cache.rcvid,
                        (fsl_size_t)size,
                        (fsl_id_t)rid);
    if(rc) goto end;
    rc = fsl_stmt_bind_blob(&s1, 1, cmpr.mem, (fsl_int_t)cmpr.used, 0);
    if(!rc){
      if(FSL_RC_STEP_DONE==fsl_stmt_step(&s1)){
        rc = fsl_db_exec(dbR, "DELETE FROM phantom "
                         "WHERE rid=%"FSL_ID_T_PFMT, (fsl_id_t)rid);
        /* MISSING: is_available() */
        if( !rc && (srcId==0 || content_is_available(srcId)) ){
          isDephantomize = 1;
          rc = fsl_content_mark_available(rid);
        }
      }
    }
#endif
  }else{
    /* We are creating a new entry */
    rc = fsl_db_prepare(dbR, &s1,
                          "INSERT INTO blob(rcvid,size,uuid,content)"
                          "VALUES("
                          "%"FSL_ID_T_PFMT","
                          "%"FSL_SIZE_T_PFMT","
                          "'%b',"
                          "?)",
                        (fsl_id_t)f->cache.rcvId,
                        (fsl_size_t)size,
                        &hash);
    if(rc) goto end;
    rc = fsl_stmt_bind_blob(&s1, 1, cmpr.mem, (fsl_int_t)cmpr.used, 0);
    if(!rc){
      rc = fsl_stmt_step(&s1);
      if(FSL_RC_STEP_DONE==rc) rc = 0;
      else goto end;
      rid = fsl_db_last_insert_id(dbR);
      if( !rc && !pBlob ){
        rc = fsl_db_exec_multi(dbR,
                               "INSERT OR IGNORE INTO phantom "
                               "VALUES(%"FSL_ID_T_PFMT")",
                               (fsl_id_t)rid);
      }
      if( !rc && (f->cache.markPrivate || isPrivate) ){
        rc = fsl_db_exec_multi(dbR,
                               "INSERT INTO private "
                               "VALUES(%"FSL_ID_T_PFMT")",
                               (fsl_id_t)rid);
        markAsUnclustered = 0;
      }
    }
    fsl_stmt_finalize(&s1);
    if(rc) goto end;
  }  

  /* If the srcId is specified, then the data we just added is
  ** really a delta.  Record this fact in the delta table.
  */
  if( srcId ){
    rc = fsl_db_exec_multi(dbR,
                           "REPLACE INTO delta(rid,srcid) VALUES("
                           "%"FSL_ID_T_PFMT","
                           "%"FSL_ID_T_PFMT")",
                           (fsl_id_t)rid, (fsl_id_t)srcId);
    if(rc) goto end;
  }
  if( !isDephantomize
      && fsl_id_bag_contains(&f->cache.arty.missing, rid) && 
      (srcId==0 || (0==fsl_acache_check_available(f,srcId)))){
    rc = fsl_content_mark_available(f, rid);
    if(rc) goto end;
  }
  if( isDephantomize ){
#if 0
    /* MISSING */
    after_dephantomize(rid, 0);
#endif
  }

  /* Add the element to the unclustered table if has never been
  ** previously seen.
  */
  if( markAsUnclustered ){
    rc = fsl_db_exec_multi(dbR,
                           "INSERT OR IGNORE INTO unclustered VALUES"
                           "(%"FSL_ID_T_PFMT")", (fsl_id_t)rid);
    if(rc) goto end;
  }

  if( markAsUnsent ){
    /* TODO: can the insertion into the UNSENT table
       reasonably happen here, perhaps with a context
       flag telling us whether we're already importing
       remote content (in which case do not insert it
       into UNSET)?
    */
    rc = fsl_db_exec(dbR, "INSERT OR IGNORE INTO unsent "
                     "VALUES(%"FSL_ID_T_PFMT")", (fsl_id_t)rid);
    if(rc) goto end;
  }
  
  rc = fsl_repo_verify_before_commit(f, rid);
  if(rc) goto end /* FSL_RC_OOM is the only possible failure here at
                     this point. */;
  
  /* Code after end: relies on the following 3 lines: */
  fsl_db_transaction_end(dbR, 0);
  inTrans = 0;
  rc = 0;
  
  if(outRid) *outRid = rid;
  end:
  if(inTrans){
    assert(0!=rc);
    fsl_db_transaction_end(dbR,1);
  }
  fsl_buffer_clear(&hash);
  if(!uncompSize){
    fsl_buffer_clear(&cmpr);
  }/* else cmpr.mem (if any) belongs to pBlob */
  return rc;
}


char fsl_acache_expire_oldest(fsl_acache * c){
  fsl_int_t i;
  fsl_int_t mnAge = c->nextAge;
  fsl_int_t mn = -1;
  for(i=0; i<c->used; i++){
    if( c->list[i].age<mnAge ){
      mnAge = c->list[i].age;
      mn = i;
    }
  }
  if( mn>=0 ){
    fsl_id_bag_remove(&c->inCache, c->list[mn].rid);
    c->szTotal -= (fsl_int_t)c->list[mn].content.used;
    fsl_buffer_clear(&c->list[mn].content);
    --c->used;
    c->list[mn] = c->list[c->used];
  }
  return (mn>=0) ? 1 : 0;
}

int fsl_acache_insert(fsl_acache * c, fsl_id_t rid, fsl_buffer *pBlob){
  static const fsl_int_t memLimit = 50000000 /* historical value */;
  static const fsl_size_t countLimit = 500 /* historical value */;
  fsl_acache_line *p;
  if( c->used>countLimit || c->szTotal>memLimit ){
    fsl_int_t szBefore;
    do{
      szBefore = c->szTotal;
      fsl_acache_expire_oldest(c);
    }while( c->szTotal>memLimit && c->szTotal<szBefore );
  }
  if( c->used>=c->capacity ){
    fsl_size_t const cap = c->capacity ? (c->capacity*2) : 10;
    void * remem = fsl_realloc(c->list, cap*sizeof(c->list[0]));
    if(!remem){
      fsl_buffer_clear(pBlob) /* for consistency */;
      return FSL_RC_OOM;
    }
    c->capacity = cap;
    c->list = (fsl_acache_line*)remem;
  }
  p = &c->list[c->used++];
  p->rid = rid;
  p->age = c->nextAge++;
  c->szTotal += pBlob->used;
  p->content = *pBlob /* Transfer ownership */;
  *pBlob = fsl_buffer_empty;
  return fsl_id_bag_insert(&c->inCache, rid);
}


void fsl_acache_clear(fsl_acache * c){
#if 0
  while(fsl_acache_expire_oldest(c)){}
#else
  fsl_size_t i;
  for(i=0; i<c->used; i++){
    fsl_buffer_clear(&c->list[i].content);
  }
#endif
  fsl_free(c->list);
  fsl_id_bag_clear(&c->missing);
  fsl_id_bag_clear(&c->available);
  fsl_id_bag_clear(&c->inCache);
  *c = fsl_acache_empty;
}


int fsl_acache_check_available(fsl_cx * f, fsl_id_t rid){
  fsl_id_t srcid;
  int depth = 0;  /* Limit to recursion depth */
  static const int limit = 10000000 /* historical value */;
  int rc;
  fsl_acache * c = &f->cache.arty;
  assert(f);
  assert(c);
  assert(rid>0);
  assert(fsl_cx_db_repo(f));
  while( depth++ < limit ){  
    fsl_int_t cSize = -1;
    if( fsl_id_bag_contains(&c->missing, rid) ){
      return FSL_RC_NOT_FOUND;
    }
    else if( fsl_id_bag_contains(&c->available, rid) ){
      return 0;
    }
    else if( (cSize=fsl_content_size(f, rid)) <0){
      rc = fsl_id_bag_insert(&c->missing, rid);
      return rc ? rc : FSL_RC_NOT_FOUND;
    }
    srcid = 0;
    rc = fsl_delta_src_id(f, rid, &srcid);
    if(rc) return rc;
    else if( srcid==0 ){
      rc = fsl_id_bag_insert(&c->available, rid);
      return rc ? rc : 0;
    }
    rid = srcid;
  }
  assert(!"delta-loop in repository");
  return fsl_cx_err_set(f, FSL_RC_CONSISTENCY,
                        "Serious problem: delta-loop in repository");
}

int fsl_content_put( fsl_cx * f, fsl_buffer const * pBlob, fsl_id_t * newRid){
  return fsl_content_put_ex(f, pBlob, NULL, 0, 0, 0, newRid);
}

char fsl_uuid_is_shunned(fsl_cx * f, fsl_uuid_cstr zUuid){
  fsl_int32_t i = 0;
  fsl_db * db = fsl_cx_db_repo(f);
  if( !db || zUuid==0 || zUuid[0]==0 ) return 0;
  i = fsl_db_g_int32( db, 0,
                      "SELECT 1 FROM shun WHERE uuid=%Q",
                      zUuid);
  return 1==i;
}


int fsl_content_new( fsl_cx * f, fsl_uuid_cstr uuid, char isPrivate,
                     fsl_id_t * newId ){
  fsl_id_t rid;
  int rc;
  fsl_db * db = fsl_cx_db_repo(f);
  fsl_stmt * s1 = NULL, * s2 = NULL;
  if(!f || !uuid) return FSL_RC_MISUSE;
  else if(!fsl_is_uuid(uuid)) return FSL_RC_RANGE;
  if(!db) return FSL_RC_NOT_A_REPO;
  if( fsl_uuid_is_shunned(f, uuid) ) return FSL_RC_ACCESS
    /* need new error code? */;
  rc = fsl_db_transaction_begin(db);
  if(rc) return rc;

  rc = fsl_db_prepare_cached(db, &s1,
                             "INSERT INTO blob(rcvid,size,uuid,content)"
                             "VALUES(0,-1,?,NULL)");
  if(rc) goto end;
  rc = fsl_stmt_bind_text(s1, 1, uuid, FSL_UUID_STRLEN, 0);
  if(!rc) rc = fsl_stmt_step(s1);
  fsl_stmt_cached_yield(s1);
  if(FSL_RC_STEP_DONE!=rc) goto end;
  else rc = 0;
  rid = fsl_db_last_insert_id(db);
  assert(rid>0);
  rc = fsl_db_prepare_cached(db, &s2,
                             "INSERT INTO phantom VALUES (?)");
  if(rc) goto end;
  rc = fsl_stmt_bind_id(s2, 1, rid);
  if(!rc) rc = fsl_stmt_step(s2);
  fsl_stmt_cached_yield(s2);
  if(FSL_RC_STEP_DONE!=rc) goto end;
  else rc = 0;

  if( f->cache.markPrivate || isPrivate ){
    /* Should be seldom enough that we don't need to cache
       this statement. */
    rc = fsl_db_exec(db,
                     "INSERT INTO private VALUES(%"FSL_ID_T_PFMT")",
                     (fsl_id_t)rid);
  }else{
    fsl_stmt * s3 = NULL;
    rc = fsl_db_prepare_cached(db, &s3,
                               "INSERT INTO unclustered VALUES(?)");
    if(!rc){
      rc = fsl_stmt_bind_id(s3, 1, rid);
      if(!rc) rc = fsl_stmt_step(s3);
      fsl_stmt_cached_yield(s3);
      if(FSL_RC_STEP_DONE!=rc) goto end;
      else rc = 0;
    }
  }

  if(!rc) rc = fsl_id_bag_insert(&f->cache.arty.missing, rid);
  
  end:
  if(rc) fsl_db_transaction_rollback(db);
  else{
    rc = fsl_db_transaction_commit(db);
    if(!rc && newId) *newId = rid;
  }
  return rc;
}

int fsl_content_undelta(fsl_cx * f, fsl_id_t rid){
  int rc;
  fsl_db * db = f ? fsl_cx_db_repo(f) : NULL;
  fsl_id_t srcid = 0;
  if(!f) return FSL_RC_MISUSE;
  else if(!db) return FSL_RC_NOT_A_REPO;
  else if(rid<=0) return FSL_RC_RANGE;
  rc = fsl_delta_src_id( f, rid, &srcid );
  if(!rc){
    fsl_buffer x = fsl_buffer_empty;
    assert(srcid>0);
    rc = fsl_content_get(f, rid, &x);
    if( !rc && x.used ){
      fsl_stmt s = fsl_stmt_empty;
      rc = fsl_db_transaction_begin(db)
        /* Reminder: the original impl does not do this in a
           transaction, _possibly_ because it's only done from places
           where a transaction is active (that's unconfirmed). If
           problems arise, try removing this transaction begin/end.
        */;
      if(!rc){
        rc = fsl_db_prepare(db, &s,
                            "UPDATE blob SET content=?,"
                            " size=%"FSL_SIZE_T_PFMT
                            " WHERE rid=%"FSL_ID_T_PFMT,
                            (fsl_size_t)x.used, (fsl_id_t)rid);
        if(!rc){
          rc = fsl_buffer_compress(&x, &x);
          if(!rc){
            rc = fsl_stmt_bind_blob(&s, 1, x.mem,
                                    (fsl_int_t)x.used, 0);
            if(!rc){
              rc = fsl_stmt_step(&s);
              if(FSL_RC_STEP_DONE==rc) rc = 0;
            }
          }
          fsl_stmt_finalize(&s);
        }
        if(!rc) rc = fsl_db_exec_multi(db,
                                       "DELETE FROM delta "
                                       "WHERE rid=%"FSL_ID_T_PFMT,
                                       (fsl_id_t)rid);
        if(rc) fsl_db_transaction_rollback(db);
        else rc = fsl_db_transaction_commit(db);
      }
    }
    fsl_buffer_clear(&x);
#if 0
    /*
      v1 does not do this, but that seems like an inconsistency.

      On that topic Richard says:

      "When you undelta an artifact, however, it is then stored as
      plain text.  (Actually, as zlib compressed plain text.)  There
      is no possibility of delta loops or bugs in the delta encoder or
      missing source artifacts.  And so there is much less of a chance
      of losing content.  Hence, I didn't see the need to verify the
      content of artifacts that are undelta-ed."

      Potential TODO: f->flags FSL_CX_F_PEDANTIC_VERIFICATION, which
      enables the R-card and this check, and any similarly superfluous
      ones.
    */
    if(!rc) fsl_repo_verify_before_commit(f, rid);
#endif
  }
  return rc;
}

int fsl_content_deltify(fsl_cx * f, fsl_id_t rid,
                             fsl_id_t srcid, char force){
  fsl_id_t s;
  fsl_buffer data = fsl_buffer_empty;
  fsl_buffer src = fsl_buffer_empty;
  fsl_buffer delta = fsl_buffer_empty;
  fsl_stmt s1 = fsl_stmt_empty;
  fsl_stmt s2 = fsl_stmt_empty;
  fsl_db * db = f ? fsl_cx_db_repo(f) : NULL;
  int rc = 0;
  if(!f) return FSL_RC_MISUSE;
  else if(rid<=0 || srcid<=0) return FSL_RC_RANGE;
  else if(!db) return FSL_RC_NOT_A_REPO;
  else if( srcid==rid ) return 0;

  if(!force){
    fsl_id_t tmpRid = 0;
    rc = fsl_delta_src_id(f, rid, &tmpRid);
    if(tmpRid>0){
      /*
        We already have a delta, it seems. Nothing left to do
        :-D. Should we return FSL_RC_ALREADY_EXISTS here?
      */
      return 0;
    }
    else if(rc) return rc;
  }

  if( fsl_content_is_private(f, srcid)
      && !fsl_content_is_private(f, rid) ){
    /*
      See API doc comments about crossing the private/public
      boundaries. Do we want to report okay here or
      FSL_RC_ACCESS? Not yet sure how this routine is used.
    */
    MARKER(("REMINDER: arguable return of 0 here.\n"));
    return 0;
  }
  s = srcid;
  while( (0==(rc=fsl_delta_src_id(f, s, &s)))
         && (s>0) ){
    if( s==rid ){
      rc = fsl_content_undelta(f, srcid);
      break;
    }
  }
  if(rc) return rc;
  /* As of here, don't return on error. Use (goto end) instead, b/c
     buffers might need cleaning.
  */
  rc = fsl_content_get(f, srcid, &src);
  if(rc) goto end;
  if( src.used < 50 ){
    /* See API doc comments about minimum size to delta/undelta. */
    fsl_buffer_clear(&src);
    return 0;
  }
  rc = fsl_content_get(f, rid, &data);
  if(rc) goto end;
  if( data.used<50 ){
    fsl_buffer_clear(&src);
    fsl_buffer_clear(&data);
    return 0;
  }
  rc = fsl_buffer_delta_create(&src, &data, &delta);
  if( !rc && (delta.used <= (data.used * 3 / 4 /* 75% */))){
    rc = fsl_buffer_compress(&delta, &delta);
    if(rc) goto end;
    /* TODO: refactor to use cached statements. */
    rc = fsl_db_prepare(db, &s1,
                        "UPDATE blob SET content=? "
                        "WHERE rid=%"FSL_ID_T_PFMT, (fsl_id_t)rid);
    if(!rc){
      rc = fsl_stmt_bind_blob(&s1, 1, delta.mem, (fsl_int_t)delta.used, 0);
      if(!rc){
        rc = fsl_db_prepare(db, &s2,
                            "REPLACE INTO delta(rid,srcid) "
                            "VALUES(%"FSL_ID_T_PFMT",%"FSL_ID_T_PFMT")",
                            (fsl_id_t)rid, (fsl_id_t)srcid);
        if(!rc){
          rc = fsl_db_transaction_begin(db);
          if(!rc){
            rc = fsl_stmt_step(&s1);
            if(FSL_RC_STEP_DONE==rc){
              rc = fsl_stmt_step(&s2);
              if(FSL_RC_STEP_DONE==rc) rc = 0;
            }
            if(!rc) rc = fsl_db_transaction_end(db, 0);
            else fsl_db_transaction_end(db, 1) /* keep rc intact */;
          }
        }
      }
    }
    fsl_stmt_finalize(&s1);
    fsl_stmt_finalize(&s2);
    if(!rc) fsl_repo_verify_before_commit(f, rid);
  }
  end:
  if(rc && db->error.code && !f->error.code){
    fsl_cx_uplift_db_error(f,db);
  }
  fsl_buffer_clear(&src);
  fsl_buffer_clear(&data);
  fsl_buffer_clear(&delta);
  return rc;
}

/**
 ** Removes all entries from the repo's blob table which are listed in
 ** the shun table.
 */
int fsl_repo_shun_artifacts(fsl_cx * f){
  fsl_stmt q = fsl_stmt_empty;
  int rc;
  fsl_db * db = f ? fsl_cx_db_repo(f) : NULL;
  if(!f) return FSL_RC_MISUSE;
  else if(!db) return FSL_RC_NOT_A_REPO;
  rc = fsl_db_exec_multi(db,
                         "CREATE TEMP TABLE IF NOT EXISTS "
                         "toshun(rid INTEGER PRIMARY KEY);"
                         "INSERT INTO toshun SELECT rid FROM blob, shun "
                         "WHERE blob.uuid=shun.uuid;"
  );
  if(rc) goto end;
  rc = fsl_db_prepare(db, &q,
                      "SELECT rid FROM delta WHERE srcid IN toshun"
                      );
  if(rc) goto end;
  while( !rc && (FSL_RC_STEP_ROW==fsl_stmt_step(&q)) ){
    fsl_id_t const srcid = fsl_stmt_g_id(&q, 0);
    rc = fsl_content_undelta(f, srcid);
  }
  fsl_stmt_finalize(&q);
  if(!rc){
    rc = fsl_db_exec_multi(db,
            "DELETE FROM delta WHERE rid IN toshun;"
            "DELETE FROM blob WHERE rid IN toshun;"
            "DROP TABLE toshun;"
            "DELETE FROM private "
            "WHERE NOT EXISTS "
            "(SELECT 1 FROM blob WHERE rid=private.rid);"
    );
  }
  end:
  if(db->error.code){
    rc = fsl_cx_err_set_e(f, &db->error);
  }
  return rc;
}

int fsl_content_make_public(fsl_cx * f, fsl_id_t rid){
  fsl_stmt * q = NULL;
  int rc;
  fsl_db * db = f ? fsl_cx_db_repo(f) : NULL;
  if(!f) return FSL_RC_MISUSE;
  else if(!db) return FSL_RC_NOT_A_REPO;
  rc = fsl_db_prepare_cached(db, &q, "DELETE FROM private WHERE rid=?");
  if(!rc){
    rc = fsl_stmt_bind_id(q, 1, rid);
    if(!rc){
      rc = fsl_stmt_step(q);
      if(FSL_RC_STEP_DONE==rc) rc = 0;
    }
    fsl_stmt_cached_yield(q);
  }
  return rc;
}

#undef MARKER