Identity database schema and publishing algorithm (#374)

I've optimized for:
- More expensive writes in exchange for more performant reads
- Support for high concurrency/reentrancy
- Consistency and correctness

README.md

@@ -33,6 +33,10 @@ You must have the _exact_ go version listed in `go.mod` - you can verify this by
 
 1. `dev/migrate-message $MIGRATION_NAME`
 
+### Create a migration for the MLS database
+
+1. `dev/migrate-mls $MIGRATION_NAME`
+
 ### Create a migration for the authz database
 
 1. `dev/migrate-authz $MIGRATION_NAME`

pkg/api/message/v1/service.go

@@ -324,9 +324,12 @@ func (s *Service) SubscribeAll(req *proto.SubscribeAllRequest, stream proto.Mess
 }
 
 func (s *Service) Query(ctx context.Context, req *proto.QueryRequest) (*proto.QueryResponse, error) {
-	log := s.log.Named("query").With(zap.Strings("content_topics", req.ContentTopics))
-	log.Debug("received request")
+	log := s.log.Named("query")
 	numContentTopics := len(req.ContentTopics)
+	if numContentTopics < 200 {
+		log = log.With(zap.Strings("content_topics", req.ContentTopics))
+	}
+	log.Debug("received request")
 
 	if numContentTopics == 0 {
 		return nil, status.Errorf(codes.InvalidArgument, "content topics required")

pkg/identity/api/v1/service.go

@@ -41,14 +41,41 @@ func (s *Service) Close() {
 	s.log.Info("closed")
 }
 
+/*
+Algorithm:
+
+Start transaction
+
+ 1. Insert the update into the inbox_log table
+ 2. Read the log for the inbox_id, ordering by sequence_id
+ 3. If the log has more than 256 entries, abort the transaction.
+ 3. Validate it sequentially. If failed, abort the transaction.
+ 4. For each affected address:
+    a. Insert or update the record with (address, inbox_id) into
+    the address_log table. Update the sequence_id if it is
+    higher
+
+End transaction
+*/
 func (s *Service) PublishIdentityUpdate(ctx context.Context, req *api.PublishIdentityUpdateRequest) (*api.PublishIdentityUpdateResponse, error) {
 	return nil, status.Errorf(codes.Unimplemented, "unimplemented")
 }
 
 func (s *Service) GetIdentityUpdates(ctx context.Context, req *api.GetIdentityUpdatesRequest) (*api.GetIdentityUpdatesResponse, error) {
+	/*
+		Algorithm for each request:
+		1. Query the inbox_log table for the inbox_id, ordering by sequence_id
+		2. Return all of the entries
+	*/
 	return nil, status.Errorf(codes.Unimplemented, "unimplemented")
 }
 
 func (s *Service) GetInboxIds(ctx context.Context, req *api.GetInboxIdsRequest) (*api.GetInboxIdsResponse, error) {
+	/*
+		Algorithm for each request:
+		1. Query the address_log table for the largest association_sequence_id
+		   for the address where revocation_sequence_id is lower or NULL
+		2. Return the value of the 'inbox_id' column
+	*/
 	return nil, status.Errorf(codes.Unimplemented, "unimplemented")
 }

pkg/migrations/mls/20240411200242_init-identity.down.sql

@@ -0,0 +1,9 @@
+SET statement_timeout = 0;
+
+--bun:split
+
+DROP TABLE IF EXISTS inbox_log;
+
+--bun:split
+
+DROP TABLE IF EXISTS address_log;

pkg/migrations/mls/20240411200242_init-identity.up.sql

@@ -0,0 +1,27 @@
+SET statement_timeout = 0;
+
+--bun:split
+
+CREATE TABLE inbox_log (
+    sequence_id BIGSERIAL PRIMARY KEY,
+    inbox_id TEXT NOT NULL,
+    server_timestamp_ns BIGINT NOT NULL,
+    identity_update_proto BYTEA NOT NULL
+);
+
+--bun:split
+
+CREATE INDEX idx_inbox_log_inbox_id ON inbox_log(inbox_id);
+
+--bun:split
+
+CREATE TABLE address_log (
+    address TEXT NOT NULL,
+    inbox_id TEXT NOT NULL,
+    association_sequence_id BIGINT,
+    revocation_sequence_id BIGINT
+);
+
+--bun:split
+
+CREATE INDEX idx_address_log_address_inbox_id ON address_log(address, inbox_id);