feat: don't store duplicated message columns (#581)

- channel_id/topic/sortkey_timestamp (in the row key)
- expiry (in the cell timestamps)

(we can always add these back to messages later if they are worth
having)

- make add_user conditional, erroring out if the user already exists
- utilize row_limit in message reads
- enforce non null columns on deserialization

Closes: SYNC-4070
Closes: SYNC-4099
Closes: SYNC-4106

Author: pjenvey

autopush-common/src/db/bigtable/bigtable_client/merge.rs

@@ -376,7 +376,6 @@ impl RowMerger {
     /// Iterate through all the returned chunks and compile them into a hash of finished cells indexed by row_key
     pub async fn process_chunks(
         mut stream: ClientSStreamReceiver<ReadRowsResponse>,
-        limit: Option<usize>,
     ) -> Result<BTreeMap<RowKey, Row>, BigTableError> {
         // Work object
         let mut merger = Self::default();
@@ -385,11 +384,6 @@ impl RowMerger {
         let mut rows = BTreeMap::<RowKey, Row>::new();
 
         while let (Some(row_resp_res), s) = stream.into_future().await {
-            if let Some(limit) = limit {
-                if limit > 0 && rows.len() > limit {
-                    break;
-                }
-            }
             stream = s;
             let row = match row_resp_res {
                 Ok(v) => v,