Skip to content

Commit

Permalink
conversations.c: Add nanosecond-based JMAP ID to G key value
Browse files Browse the repository at this point in the history 
Also, create a new I key which maps JMAP ID to GUID (G key)
  • Loading branch information
@ksmurchison
ksmurchison committed Jan 6, 2025
1 parent 6a23b92 commit 4fb40bd
Show file tree
Hide file tree
Showing 3 changed files with 108 additions and 39 deletions.
114 changes: 87 additions & 27 deletions imap/conversations.c
View file
Original file line number Diff line number Diff line change
Expand Up @@ -1964,6 +1964,7 @@ static int _guid_one(struct guid_foreach_rock *frock,
uint32_t system_flags,
uint32_t internal_flags,
time_t internaldate,
const char *jidrep,
char version)
{
const char *p, *err;
Expand All @@ -1980,6 +1981,8 @@ static int _guid_one(struct guid_foreach_rock *frock,
rec.internaldate = internaldate;
rec.version = version;

strlcpy(rec.jidrep, jidrep, sizeof(rec.jidrep));

/* ensure a NULL terminated key string */
buf_cstring(&frock->partbuf);
char *item = frock->partbuf.s;
Expand Down Expand Up @@ -2056,7 +2059,7 @@ static int _guid_cb(void *rock,
buf_setcstr(&frock->partbuf, strarray_nth(recs, i));
r = _guid_one(frock, key, /*cid*/0, /*basecid*/0,
/*system_flags*/0, /*internal_flags*/0,
/*internaldate*/0, /*version*/0);
/*internaldate*/0, /*jidrep*/0, /*version*/0);
if (r) break;
}
strarray_free(recs);
Expand All @@ -2072,6 +2075,7 @@ static int _guid_cb(void *rock,
uint32_t system_flags = 0;
uint32_t internal_flags = 0;
time_t internaldate = 0;
char jidrep[CONV_JMAPID_LEN+1] = "";
char version = 0;
if (datalen >= 16) {
const char *p = data;
Expand Down Expand Up @@ -2103,6 +2107,24 @@ static int _guid_cb(void *rock,
p += 8;
break;

case 3:
/* cid */
cid = ntohll(*((bit64*)p));
p += 8;
/* system_flags */
system_flags = ntohl(*((bit32*)p));
p += 4;
/* internal flags */
internal_flags = ntohl(*((bit32*)p));
p += 4;
/* internaldate*/
internaldate = (time_t) ntohll(*((bit64*)p));
p += 8;
/* basecid */
basecid = ntohll(*((bit64*)p));
p += 8;
break;

default:
/* cid */
cid = ntohll(*((bit64*)p));
Expand All @@ -2119,13 +2141,17 @@ static int _guid_cb(void *rock,
/* basecid */
basecid = ntohll(*((bit64*)p));
p += 8;
/* emailid */
memcpy(jidrep, p, CONV_JMAPID_LEN);
jidrep[CONV_JMAPID_LEN] = '\0';
p += CONV_JMAPID_LEN;
break;
}
}

buf_setmap(&frock->partbuf, key+42, keylen-42);
r = _guid_one(frock, key, cid, basecid, system_flags, internal_flags,
internaldate, version);
internaldate, jidrep, version);

return r;
}
Expand Down Expand Up @@ -2229,40 +2255,58 @@ static int conversations_guid_setitem(struct conversations_state *state,
uint32_t system_flags,
uint32_t internal_flags,
time_t internaldate,
bit64 jmapid,
int add)
{
struct buf key = BUF_INITIALIZER;
buf_setcstr(&key, "G");
buf_appendcstr(&key, guidrep);
struct buf gkey = BUF_INITIALIZER;
struct buf ikey = BUF_INITIALIZER;
buf_setcstr(&gkey, "G");
buf_appendcstr(&gkey, guidrep);
size_t datalen = 0;
const char *data;

// check if we have to upgrade anything?
int r = cyrusdb_fetch(state->db, buf_base(&key), buf_len(&key), &data, &datalen, &state->txn);
int r = cyrusdb_fetch(state->db, buf_base(&gkey), buf_len(&gkey), &data, &datalen, &state->txn);
if (!r && datalen) {
int i;
buf_putc(&key, ':');
buf_putc(&gkey, ':');

/* add new keys for all the old values */
strarray_t *old = strarray_nsplit(data, datalen, ",", /*flags*/0);
for (i = 0; i < strarray_size(old); i++) {
buf_truncate(&key, 42); // trim back to the colon
buf_appendcstr(&key, strarray_nth(old, i));
r = cyrusdb_store(state->db, buf_base(&key), buf_len(&key), "", 0, &state->txn);
buf_truncate(&gkey, 42); // trim back to the colon
buf_appendcstr(&gkey, strarray_nth(old, i));
r = cyrusdb_store(state->db, buf_base(&gkey), buf_len(&gkey), "", 0, &state->txn);
if (r) break;
}
strarray_free(old);
if (r) goto done;

buf_truncate(&key, 41); // trim back to original key
buf_truncate(&gkey, 41); // trim back to original key

/* remove the original key */
r = cyrusdb_delete(state->db, buf_base(&key), buf_len(&key), &state->txn, /*force*/0);
r = cyrusdb_delete(state->db, buf_base(&gkey), buf_len(&gkey), &state->txn, /*force*/0);
if (r) goto done;
}

buf_putc(&key, ':');
buf_appendcstr(&key, item);
buf_putc(&gkey, ':');
buf_appendcstr(&gkey, item);

/* Build I key */
buf_setcstr(&ikey, "I");
if (!jmapid) {
/* Updating an existing record - lookup jmapidrep */
r = cyrusdb_fetch(state->db, buf_base(&gkey), buf_len(&gkey),
&data, &datalen, &state->txn);
if (!r && datalen) {
buf_appendmap(&ikey, data + CONV_JMAPID_OFFSET, CONV_JMAPID_LEN);
}
}
else {
/* Creating a new record - encode given jmapid */
bit64 u64 = htonll(UINT64_MAX - jmapid);
charset_encode(&ikey, (const char *) &u64, 8, ENCODING_BASE64JMAPID);
}

if (add) {
struct buf val = BUF_INITIALIZER;
Expand All @@ -2281,19 +2325,33 @@ static int conversations_guid_setitem(struct conversations_state *state,
buf_appendbit32(&val, internal_flags);
buf_appendbit64(&val, (bit64)internaldate);
buf_appendbit64(&val, basecid == cid ? 0 : basecid);
buf_appendmap(&val, buf_base(&ikey) + 1, CONV_JMAPID_LEN);
}

r = cyrusdb_store(state->db, buf_base(&key), buf_len(&key),
r = cyrusdb_store(state->db, buf_base(&gkey), buf_len(&gkey),
buf_base(&val), buf_len(&val),
&state->txn);

if (!r && buf_len(&ikey)) {
/* Insert I record: jidrep -> guidrep */
r = cyrusdb_store(state->db, buf_base(&ikey), buf_len(&ikey),
guidrep, strlen(guidrep),
&state->txn);
}

buf_free(&val);
}
else {
r = cyrusdb_delete(state->db, buf_base(&key), buf_len(&key), &state->txn, /*force*/1);
r = cyrusdb_delete(state->db, buf_base(&gkey), buf_len(&gkey), &state->txn, /*force*/1);
if (!r && buf_len(&ikey)) {
r = cyrusdb_delete(state->db, buf_base(&ikey), buf_len(&ikey),
&state->txn, /*force*/1);
}
}

done:
buf_free(&key);
buf_free(&gkey);
buf_free(&ikey);

return r;
}
Expand All @@ -2303,6 +2361,7 @@ static int _guid_addbody(struct conversations_state *state,
conversation_id_t basecid,
uint32_t system_flags, uint32_t internal_flags,
time_t internaldate,
bit64 jmapid,
struct body *body,
const char *base, int add)
{
Expand All @@ -2319,17 +2378,17 @@ static int _guid_addbody(struct conversations_state *state,
const char *guidrep = message_guid_encode(&body->content_guid);
r = conversations_guid_setitem(state, guidrep, buf_cstring(&buf), cid, basecid,
system_flags, internal_flags, internaldate,
add);
jmapid, add);
buf_free(&buf);

if (r) return r;
}

r = _guid_addbody(state, cid, basecid, system_flags, internal_flags, internaldate, body->subpart, base, add);
r = _guid_addbody(state, cid, basecid, system_flags, internal_flags, internaldate, jmapid, body->subpart, base, add);
if (r) return r;

for (i = 1; i < body->numparts; i++) {
r = _guid_addbody(state, cid, basecid, system_flags, internal_flags, internaldate, body->subpart + i, base, add);
r = _guid_addbody(state, cid, basecid, system_flags, internal_flags, internaldate, jmapid, body->subpart + i, base, add);
if (r) return r;
}

Expand All @@ -2339,7 +2398,7 @@ static int _guid_addbody(struct conversations_state *state,
static int conversations_set_guid(struct conversations_state *state,
struct mailbox *mailbox,
const struct index_record *record,
int add)
bit64 jmapid, int add)
{
int folder = conversation_folder_number(state,
CONV_FOLDER_KEY_MBOX(state, mailbox),
Expand All @@ -2362,10 +2421,10 @@ static int conversations_set_guid(struct conversations_state *state,
record->system_flags,
record->internal_flags,
record->internaldate,
add);
jmapid, add);
if (!r) r = _guid_addbody(state, record->cid, record->basecid,
record->system_flags, record->internal_flags,
record->internaldate, body, base, add);
record->internaldate, jmapid, body, base, add);

message_free_body(body);
free(body);
Expand Down Expand Up @@ -2435,6 +2494,7 @@ EXPORTED int conversations_update_record(struct conversations_state *cstate,
struct mailbox *mailbox,
const struct index_record *old,
struct index_record *new,
bit64 jmapid,
int allowrenumber,
int ignorelimits,
int silent)
Expand Down Expand Up @@ -2463,9 +2523,9 @@ EXPORTED int conversations_update_record(struct conversations_state *cstate,
* a removal and re-add, so cache gets parsed and msgids
* updated */
if (old->cid != new->cid) {
r = conversations_update_record(cstate, mailbox, old, NULL, 0, ignorelimits, silent);
r = conversations_update_record(cstate, mailbox, old, NULL, jmapid, 0, ignorelimits, silent);
if (r) return r;
return conversations_update_record(cstate, mailbox, NULL, new, 0, ignorelimits, silent);
return conversations_update_record(cstate, mailbox, NULL, new, jmapid, 0, ignorelimits, silent);
}
}

Expand Down Expand Up @@ -2519,13 +2579,13 @@ EXPORTED int conversations_update_record(struct conversations_state *cstate,
if (!old || old->system_flags != new->system_flags ||
old->internal_flags != new->internal_flags ||
old->internaldate != new->internaldate) {
r = conversations_set_guid(cstate, mailbox, new, /*add*/1);
r = conversations_set_guid(cstate, mailbox, new, jmapid, /*add*/1);
if (r) goto done;
}
}
else {
if (old) {
r = conversations_set_guid(cstate, mailbox, old, /*add*/0);
r = conversations_set_guid(cstate, mailbox, old, jmapid, /*add*/0);
if (r) goto done;
}
}
Expand Down
9 changes: 7 additions & 2 deletions imap/conversations.h
View file
Original file line number Diff line number Diff line change
Expand Up @@ -138,21 +138,25 @@ struct conv_folder {
uint32_t prev_exists;
};

#define CONV_GUIDREC_VERSION 3 // (must be <= 127)
#define CONV_GUIDREC_VERSION 4 // (must be <= 127)
#define CONV_GUIDREC_BYNAME_VERSION 1 // last folders byname version

#define CONV_JMAPID_LEN 11 // 64 bits base64-encoded w/o padding
#define CONV_JMAPID_OFFSET 33 // offset of JMAPID in G key value (1+8+4+4+8+8)

struct conv_guidrec {
const struct conversations_state *cstate; // this conversationsdb!
const char *guidrep; // [MESSAGE_GUID_SIZE*2], hex-encoded
int foldernum;
uint32_t uid;
const char *part;
conversation_id_t cid;
conversation_id_t basecid;
conversation_id_t basecid; // if version >= 3
char version;
uint32_t system_flags; // if version >= 1
uint32_t internal_flags; // if version >= 1
time_t internaldate; // if version >= 1
char jidrep[CONV_JMAPID_LEN+1]; // if version >= 4
};

struct conv_sender {
Expand Down Expand Up @@ -329,6 +333,7 @@ extern int conversations_update_record(struct conversations_state *cstate,
struct mailbox *mailbox,
const struct index_record *old,
struct index_record *new_,
bit64 jmapid,
int allowrenumber,
int ignorelimits,
int silent);
Expand Down
24 changes: 14 additions & 10 deletions imap/mailbox.c
View file
Original file line number Diff line number Diff line change
Expand Up @@ -4392,7 +4392,8 @@ EXPORTED struct conversations_state *mailbox_get_cstate_full(struct mailbox *mai

static int mailbox_update_conversations(struct mailbox *mailbox,
const struct index_record *old,
struct index_record *new)
struct index_record *new,
bit64 jmapid)
{
struct conversations_state *cstate = mailbox_get_cstate(mailbox);

Expand All @@ -4407,7 +4408,7 @@ static int mailbox_update_conversations(struct mailbox *mailbox,
return 0;

int ignorelimits = new ? new->ignorelimits : 1;
return conversations_update_record(cstate, mailbox, old, new,
return conversations_update_record(cstate, mailbox, old, new, jmapid,
/*allowrenumber*/1, ignorelimits, mailbox->silentchanges);
}

Expand Down Expand Up @@ -4456,12 +4457,13 @@ EXPORTED int mailbox_update_xconvmodseq(struct mailbox *mailbox, modseq_t newmod
* we expect callers to have already done all sanity checking */
static int mailbox_update_indexes(struct mailbox *mailbox,
const struct index_record *old,
struct index_record *new)
struct index_record *new,
bit64 jmapid)
{
int r = 0;

// 'new' is not static because conversations might change the CID
r = mailbox_update_conversations(mailbox, old, new);
r = mailbox_update_conversations(mailbox, old, new, jmapid);
if (r) return r;

#ifdef WITH_DAV
Expand Down Expand Up @@ -4596,7 +4598,7 @@ EXPORTED int mailbox_rewrite_index_record(struct mailbox *mailbox,
if (r) return r;
}

r = mailbox_update_indexes(mailbox, &oldrecord, record);
r = mailbox_update_indexes(mailbox, &oldrecord, record, 0);
if (r) return r;

if ((record->internal_flags & FLAG_INTERNAL_EXPUNGED) && !(changeflags & CHANGE_WASEXPUNGED)) {
Expand Down Expand Up @@ -4758,7 +4760,9 @@ EXPORTED int mailbox_append_index_record(struct mailbox *mailbox,
if (r) return r;
}

r = mailbox_update_indexes(mailbox, NULL, record);
r = mailbox_update_indexes(mailbox, NULL, record,
/* jmapid is nanoseconds since epoch */
record->internaldate * 1000000000 + now.tv_nsec);
if (r) return r;

record->recno = mailbox->i.num_records + 1;
Expand Down Expand Up @@ -6193,7 +6197,7 @@ EXPORTED int mailbox_add_conversations(struct mailbox *mailbox, int silent)

struct index_record copyrecord = *record;
copyrecord.silentupdate = silent;
r = conversations_update_record(cstate, mailbox, NULL, &copyrecord, 1,
r = conversations_update_record(cstate, mailbox, NULL, &copyrecord, 0, 1,
/*ignorelimits*/1, silent);
if (r) break;

Expand All @@ -6203,15 +6207,15 @@ EXPORTED int mailbox_add_conversations(struct mailbox *mailbox, int silent)
assert(!silent); // can't change cid if silent!

/* remove this record again */
r = conversations_update_record(cstate, mailbox, &copyrecord, NULL, 0,
r = conversations_update_record(cstate, mailbox, &copyrecord, NULL, 0, 0,
/*ignorelimits*/1, silent);
if (r) break;

/* we had a cid change, so rewrite will try to correct the counts, so we
* need to add this one in again */
struct index_record oldrecord = *record;
/* add the old record that's going away */
r = conversations_update_record(cstate, mailbox, NULL, &oldrecord, 0,
r = conversations_update_record(cstate, mailbox, NULL, &oldrecord, 0, 0,
/*ignorelimits*/1, silent);
if (r) break;

Expand Down Expand Up @@ -6239,7 +6243,7 @@ static int mailbox_delete_conversations(struct mailbox *mailbox)
if (!record->cid)
continue;

r = conversations_update_record(cstate, mailbox, record, NULL,
r = conversations_update_record(cstate, mailbox, record, NULL, 0,
/*allowrenumber*/0, /*ignorelimits*/1,
mailbox->silentchanges);
if (r) break;
Expand Down

0 comments on commit 4fb40bd

Please sign in to comment.