diff options
| author | EuAndreh <eu@euandre.org> | 2025-09-20 15:45:46 -0300 |
|---|---|---|
| committer | EuAndreh <eu@euandre.org> | 2025-09-20 15:50:05 -0300 |
| commit | aee466b543b9d87a0206a59b1ade4686aa1f459e (patch) | |
| tree | 386b25e72df3e44acc052bfc2806523d190f8de3 /src | |
| parent | meta.capim: Incorporate previous long-description file (diff) | |
| download | papod-aee466b543b9d87a0206a59b1ade4686aa1f459e.tar.gz papod-aee466b543b9d87a0206a59b1ade4686aa1f459e.tar.xz | |
Remove SQLite code
Diffstat (limited to 'src')
| -rw-r--r-- | src/papod.go | 3759 |
1 files changed, 142 insertions, 3617 deletions
diff --git a/src/papod.go b/src/papod.go index b6544da..5bf2467 100644 --- a/src/papod.go +++ b/src/papod.go @@ -3,8 +3,8 @@ package papod import ( "bufio" "bytes" - "database/sql" "errors" + "flag" "fmt" "io" "log/slog" @@ -13,11 +13,9 @@ import ( "regexp" "strings" "sync" - "time" + "acude" "cracha" - "fiinha" - "golite" "uuid" "pds" "stm" @@ -27,27 +25,11 @@ import ( -const ( - defaultPrefix = "papod" - rollbackErrorFmt = "rollback error: %w; while executing: %w" - - NEW_CHANNEL = "new-channel" - - pingFrequency = time.Duration(30) * time.Second - pongMaxLatency = time.Duration(5) * time.Second -) - - - -type dbconfigT struct{ - shared *sql.DB - dbpath string - prefix string +type newUserT struct{ } -type queryT struct{ - write string - read string +type userT struct{ + username string } type queriesT struct{ @@ -55,156 +37,7 @@ type queriesT struct{ userByUUID func(uuid.UUID) (userT, error) updateUser func(userT) error deleteUser func(uuid.UUID) error - addNetwork func(userT, newNetworkT, uuid.UUID) (networkT, error) - getNetwork func(userT, uuid.UUID) (networkT, error) - networks func(userT, func(networkT) error) error - setNetwork func(memberT, networkT) error - nipNetwork func(memberT) error - membership func(userT, networkT) (memberT, error) - addMember func(memberT, newMemberT) (memberT, error) - addRole func(memberT, string, memberT) error - dropRole func(memberT, string, memberT) error - showMember func(memberT, uuid.UUID) (memberT, error) - members func(memberT, func(memberT) error) error - editMember func(memberT, memberT) error - dropMember func(memberT, uuid.UUID) error - addChannel func(memberT, newChannelT) (channelT, error) - chanByName func(memberT, string) (channelT, error) - channels func(memberT, func(channelT) error) error - setChannel func(memberT, channelT) error - endChannel func(memberT, uuid.UUID) error - join func(memberT, uuid.UUID) error - part func(memberT, channelT) error - names func(memberT, uuid.UUID, func(memberT) error) error - addEvent func(newEventT) (eventT, error) - allAfter func(memberT, uuid.UUID, func(eventT) error) error - logMessage func(userT, messageT) error - close func() error -} - -type newUserT struct{ - uuid uuid.UUID - username string - displayName string -} - -type userT struct{ - id int64 - timestamp time.Time - uuid uuid.UUID - username string - displayName string - pictureID *uuid.UUID -} - -type NetworkType string -const ( - NetworkType_Public NetworkType = "public" - NetworkType_Private NetworkType = "private" - NetworkType_Unlisted NetworkType = "unlisted" -) - -type MemberStatus string -const ( - MemberStatus_Active MemberStatus = "active" - MemberStatus_Inactive MemberStatus = "inactive" - MemberStatus_Removed MemberStatus = "removed" -) - -type newNetworkT struct{ - uuid uuid.UUID - name string - description string - type_ NetworkType -} - -type networkT struct{ - id int64 - timestamp time.Time - uuid uuid.UUID - name string - description string - type_ NetworkType -} - -type newMemberT struct{ - userID uuid.UUID - memberID uuid.UUID - username string -} - -type memberT struct{ - id int64 - timestamp time.Time - uuid uuid.UUID - username string - displayName string - pictureID *uuid.UUID - status MemberStatus - roles []string -} - -type newChannelT struct{ - id int64 - timestamp time.Time - uuid uuid.UUID - publicName *string - label string - description string - virtual bool -} - -type channelT struct{ - id int64 - timestamp time.Time - uuid uuid.UUID - publicName *string - label string - description string - virtual bool -} - -type SourceType string -const ( - SourceType_Logon SourceType = "logon" -) - -type sourceT struct{ - uuid uuid.UUID - type_ SourceType - metadata *map[string]interface{} -} - -type EventType string -const ( - EventType_UserJoin EventType = "user-join" - EventType_UserMessage EventType = "user-message" -) - -type newEventT struct{ - eventID uuid.UUID - channelID uuid.UUID - source sourceT - type_ EventType - payload string - metadata *map[string]interface{} -} - -type eventT struct{ - id int64 - timestamp time.Time - uuid uuid.UUID - channelID uuid.UUID - source sourceT - type_ EventType - payload string - metadata *map[string]interface{} -} - -type eventEntryT struct{ - event eventT - previous *eventT - isFirst bool + createNetwork func(userT, newNetworkT, uuid.UUID) (networkT, error) } type messageT struct{ @@ -219,19 +52,22 @@ type replyT struct{ params []string } +type taskT struct{ +} + +type actionT struct{ + replies []replyT + shouldClose bool + err error + task []taskT +} + type listenersT struct{ daemon net.Listener commander net.Listener close func() error } -type consumerT struct{ - topic string - name string - // FIXME: use generic to avoid circular reference? - handlerFn func(papodT) func(fiinha.Message) error -} - type netConnI interface{ Write(p []byte) (n int, err error) Close() error @@ -252,3264 +88,48 @@ type metricsT struct{ sentReply func(...any) } -type stateMutableT struct{ - // disconnect func(*connectionT) - subscribe func(string, []string) -} - type stateT struct{ members *pds.Map[string, []string] users *pds.Map[string, []uuid.UUID] connections *pds.Map[uuid.UUID, connectionT] } -// TODO: key for members should be the channelID, not its name -type stateMutableDataT struct{ - connections map[uuid.UUID]*connectionT - users map[string][]uuid.UUID - members map[string]map[string][]uuid.UUID -} - type papodT struct{ - auth cracha.IAuth - queue fiinha.IQueue - queries queriesT + acude acude.AcudeI + cracha cracha.CrachaI listeners listenersT - consumers []consumerT - stateMutable stateMutableT - state *stm.AtomT[*stateT] + state *stm.AtomT[stateT] metrics metricsT - // logger g.Logger } -type IPapod interface{ +type PapoI interface{ Start() error Close() error } - - -func serialized[A any, B any](callback func(...A) B) (func(...A) B, func()) { - in := make(chan []A) - out := make(chan B) - - closed := false - var ( - closeWg sync.WaitGroup - closeMutex sync.Mutex - ) - closeWg.Add(1) - - go func() { - for input := range in { - out <- callback(input...) - } - close(out) - closeWg.Done() - }() - - fn := func(input ...A) B { - in <- input - return (<- out) - } - - closeFn := func() { - closeMutex.Lock() - defer closeMutex.Unlock() - if closed { - return - } - close(in) - closed = true - closeWg.Wait() - } - - return fn, closeFn -} - -func execSerialized(query string, db *sql.DB) (func(...any) error, func()) { - return serialized(func(args ...any) error { - return inTx(db, func(tx *sql.Tx) error { - _, err := tx.Exec(query, args...) - return err - }) - }) -} - -func tryRollback(tx *sql.Tx, err error) error { - rollbackErr := tx.Rollback() - if rollbackErr != nil { - return fmt.Errorf( - rollbackErrorFmt, - rollbackErr, - err, - ) - } - - return err +type argsT struct{ + allArgs []string + subArgs []string + baseDir string + tag string } -func inTx(db *sql.DB, fn func(*sql.Tx) error) error { - tx, err := db.Begin() - if err != nil { - return err - } - - err = fn(tx) - if err != nil { - return tryRollback(tx, err) - } - - err = tx.Commit() - if err != nil { - return tryRollback(tx, err) - } - - return nil +type envT struct{ + args argsT + in io.Reader + out io.Writer + err io.Writer } -/// "papod_users".uuid is the same as cracha_users.uuid. Not a foreign key to -/// allow them to live in different physical locations. Adding it here feels -/// less like an optimization related decision, and more of a coupling one. The -/// way that New() works now uses the same databasePath for the fiinha.IQueue -/// *and* cracha.IAuth, but cracha in no way exposes where it stores the user -/// UUID or how the it is handled. This has similarities to how events here -/// don't reference the fiinha.Message.ID via foreign keys either. They're -/// treated only as opaque IDs. -func createTablesSQL(prefix string) queryT { - const tmpl_write = ` - -- TODO: unconfirmed premise: statements within a trigger are - -- part of the transaction that caused it, and so are - -- atomic. - -- See also: - -- https://stackoverflow.com/questions/77441888/ - -- https://stackoverflow.com/questions/30511116/ - - CREATE TABLE IF NOT EXISTS "%s_users" ( - id INTEGER NOT NULL PRIMARY KEY AUTOINCREMENT, - timestamp TEXT NOT NULL DEFAULT ( - %s - ), - -- provided by cracha - user_uuid BLOB NOT NULL UNIQUE, - username TEXT NOT NULL, - display_name TEXT NOT NULL, - picture_uuid BLOB UNIQUE, - deleted INT NOT NULL CHECK(deleted IN (0, 1)) - ) STRICT; - CREATE TABLE IF NOT EXISTS "%s_user_changes" ( - id INTEGER NOT NULL PRIMARY KEY AUTOINCREMENT, - timestamp TEXT NOT NULL DEFAULT ( - %s - ), - user_id INTEGER NOT NULL, - attribute TEXT NOT NULL CHECK( - attribute IN ( - 'username', - 'display_name', - 'picture_uuid', - 'deleted' - ) - ), - value_text TEXT, - value_blob BLOB, - value_bool INT CHECK(value_bool IN (0, 1)), - op INT NOT NULL CHECK(op IN (0, 1)) - ) STRICT; - CREATE TRIGGER IF NOT EXISTS "%s_user_new" - AFTER INSERT ON "%s_users" - BEGIN - INSERT INTO "%s_user_changes" ( - user_id, attribute, value_text, op - ) VALUES - (NEW.id, 'username', NEW.username, true), - (NEW.id, 'display_name', NEW.display_name, true) - ; - INSERT INTO "%s_user_changes" ( - user_id, attribute, value_bool, op - ) VALUES - (NEW.id, 'deleted', NEW.deleted, true) - ; - END; - CREATE TRIGGER IF NOT EXISTS "%s_user_new_picture_uuid" - AFTER INSERT ON "%s_users" - WHEN NEW.picture_uuid IS NOT NULL - BEGIN - INSERT INTO "%s_user_changes" ( - user_id, attribute, value_blob, op - ) VALUES - (NEW.id, 'picture_uuid', NEW.picture_uuid, true) - ; - END; - CREATE TRIGGER IF NOT EXISTS "%s_user_update_username" - AFTER UPDATE ON "%s_users" - WHEN OLD.username != NEW.username - BEGIN - INSERT INTO "%s_user_changes" ( - user_id, attribute, value_text, op - ) VALUES - (NEW.id, 'username', OLD.username, false), - (NEW.id, 'username', NEW.username, true) - ; - END; - CREATE TRIGGER IF NOT EXISTS "%s_user_update_display_name" - AFTER UPDATE ON "%s_users" - WHEN OLD.display_name != NEW.display_name - BEGIN - INSERT INTO "%s_user_changes" ( - user_id, attribute, value_text, op - ) VALUES - (NEW.id, 'display_name', OLD.display_name, false), - (NEW.id, 'display_name', NEW.display_name, true) - ; - END; - CREATE TRIGGER IF NOT EXISTS "%s_user_add_picture_uuid" - AFTER UPDATE ON "%s_users" - WHEN ( - OLD.picture_uuid IS NULL AND - NEW.picture_uuid IS NOT NULL - ) - BEGIN - INSERT INTO "%s_user_changes" ( - user_id, attribute, value_blob, op - ) VALUES - (NEW.id, 'picture_uuid', NEW.picture_uuid, true) - ; - END; - CREATE TRIGGER IF NOT EXISTS "%s_user_remove_picture_uuid" - AFTER UPDATE ON "%s_users" - WHEN ( - OLD.picture_uuid IS NOT NULL AND - NEW.picture_uuid IS NULL - ) - BEGIN - INSERT INTO "%s_user_changes" ( - user_id, attribute, value_blob, op - ) VALUES - (NEW.id, 'picture_uuid', OLD.picture_uuid, false) - ; - END; - CREATE TRIGGER IF NOT EXISTS "%s_user_update_picture_uuid" - AFTER UPDATE ON "%s_users" - WHEN ( - OLD.picture_uuid IS NOT NULL AND - NEW.picture_uuid IS NOT NULL AND - OLD.picture_uuid != NEW.picture_uuid - ) - BEGIN - INSERT INTO "%s_user_changes" ( - user_id, attribute, value_blob, op - ) VALUES - (NEW.id, 'picture_uuid', OLD.picture_uuid, false), - (NEW.id, 'picture_uuid', NEW.picture_uuid, true) - ; - END; - CREATE TRIGGER IF NOT EXISTS "%s_user_update_deleted" - AFTER UPDATE ON "%s_users" - WHEN OLD.deleted != NEW.deleted - BEGIN - INSERT INTO "%s_user_changes" ( - user_id, attribute, value_bool, op - ) VALUES - (NEW.id, 'deleted', OLD.deleted, false), - (NEW.id, 'deleted', NEW.deleted, true) - ; - END; - - CREATE TABLE IF NOT EXISTS "%s_sessions" ( - id INTEGER NOT NULL PRIMARY KEY AUTOINCREMENT, - timestamp TEXT NOT NULL DEFAULT ( - %s - ), - -- provided by cracha - session_uuid BLOB NOT NULL UNIQUE, - user_id INTEGER NOT NULL - REFERENCES "%s_users"(id), - finished_at TEXT - ); - CREATE TABLE IF NOT EXISTS "%s_connections" ( - id INTEGER NOT NULL PRIMARY KEY AUTOINCREMENT, - timestamp TEXT NOT NULL DEFAULT ( - %s - ), - uuid BLOB NOT NULL UNIQUE, - finished_at TEXT - ); - CREATE TABLE IF NOT EXISTS "%s_logons" ( - id INTEGER NOT NULL PRIMARY KEY AUTOINCREMENT, - timestamp TEXT NOT NULL DEFAULT ( - %s - ), - session_id INTEGER NOT NULL - REFERENCES "%s_sessions"(id), - connection_id INTEGER NOT NULL - REFERENCES "%s_connections"(id), - UNIQUE (session_id, connection_id) - ) STRICT; - - CREATE TABLE IF NOT EXISTS "%s_networks" ( - id INTEGER NOT NULL PRIMARY KEY AUTOINCREMENT, - timestamp TEXT NOT NULL DEFAULT ( - %s - ), - uuid BLOB NOT NULL UNIQUE, - name TEXT NOT NULL, - description TEXT NOT NULL, - type TEXT NOT NULL CHECK( - type IN ('public', 'private', 'unlisted') - ), - deleted INT NOT NULL CHECK(deleted IN (0, 1)) - ) STRICT; - CREATE INDEX IF NOT EXISTS "%s_networks_type" - ON "%s_networks"(type); - CREATE TABLE IF NOT EXISTS "%s_network_changes" ( - id INTEGER NOT NULL PRIMARY KEY AUTOINCREMENT, - timestamp TEXT NOT NULL DEFAULT ( - %s - ), - network_id INTEGER NOT NULL, - attribute TEXT NOT NULL CHECK( - attribute IN ( - 'name', - 'description', - 'type', - 'deleted', - 'logon_id' -- FIXME - ) - ), - value TEXT NOT NULL, - op INT NOT NULL CHECK(op IN (0, 1)) - ) STRICT; - CREATE TRIGGER IF NOT EXISTS "%s_network_new" - AFTER INSERT ON "%s_networks" - BEGIN - INSERT INTO "%s_network_changes" ( - network_id, attribute, value, op - ) VALUES - (NEW.id, 'name', NEW.name, true), - (NEW.id, 'description', NEW.description, true), - (NEW.id, 'type', NEW.type, true), - (NEW.id, 'deleted', NEW.deleted, true) - ; - END; - CREATE TRIGGER IF NOT EXISTS "%s_network_update_name" - AFTER UPDATE ON "%s_networks" - WHEN OLD.name != NEW.name - BEGIN - INSERT INTO "%s_network_changes" ( - network_id, attribute, value, op - ) VALUES - (NEW.id, 'name', OLD.name, false), - (NEW.id, 'name', NEW.name, true) - ; - END; - CREATE TRIGGER IF NOT EXISTS "%s_network_update_description" - AFTER UPDATE ON "%s_networks" - WHEN OLD.description != NEW.description - BEGIN - INSERT INTO "%s_network_changes" ( - network_id, attribute, value, op - ) VALUES - (NEW.id, 'description', OLD.description, false), - (NEW.id, 'description', NEW.description, true) - ; - END; - CREATE TRIGGER IF NOT EXISTS "%s_network_update_type" - AFTER UPDATE ON "%s_networks" - WHEN OLD.description != NEW.description - BEGIN - INSERT INTO "%s_network_changes" ( - network_id, attribute, value, op - ) VALUES - (NEW.id, 'type', OLD.type, false), - (NEW.id, 'type', NEW.type, true) - ; - END; - CREATE TRIGGER IF NOT EXISTS "%s_network_update_deleted" - AFTER UPDATE ON "%s_networks" - WHEN OLD.deleted != NEW.deleted - BEGIN - INSERT INTO "%s_network_changes" ( - network_id, attribute, value, op - ) VALUES - (NEW.id, 'deleted', OLD.deleted, false), - (NEW.id, 'deleted', NEW.deleted, true) - ; - END; - - CREATE TABLE IF NOT EXISTS "%s_members" ( - id INTEGER NOT NULL PRIMARY KEY AUTOINCREMENT, - timestamp TEXT NOT NULL DEFAULT ( - %s - ), - uuid BLOB NOT NULL UNIQUE, - network_id INTEGER NOT NULL - REFERENCES "%s_networks"(id), - user_id INTEGER NOT NULL, - username TEXT NOT NULL, - display_name TEXT NOT NULL, - picture_uuid BLOB UNIQUE, - status TEXT NOT NULL CHECK( - status IN ('active', 'inactive', 'removed') - ), - active_uniq TEXT CHECK( - active_uniq IN ('active', NULL) - ), - UNIQUE (network_id, username, active_uniq), - UNIQUE (network_id, user_id) - ) STRICT; - CREATE TABLE IF NOT EXISTS "%s_member_changes" ( - id INTEGER NOT NULL PRIMARY KEY AUTOINCREMENT, - timestamp TEXT NOT NULL DEFAULT ( - %s - ), - member_id INTEGER NOT NULL, - attribute TEXT NOT NULL CHECK( - attribute IN ( - 'username', - 'display_name', - 'picture_uuid', - 'status', - 'logon_id' -- FIXME - ) - ), - value_text TEXT, - value_blob BLOB, - op INT NOT NULL CHECK(op IN (0, 1)) - ) STRICT; - CREATE TRIGGER IF NOT EXISTS "%s_member_new" - AFTER INSERT ON "%s_members" - BEGIN - INSERT INTO "%s_member_changes" ( - member_id, attribute, value_text, op - ) VALUES - (NEW.id, 'username', NEW.username, true), - (NEW.id, 'display_name', NEW.display_name, true), - (NEW.id, 'status', NEW.status, true) - ; - END; - CREATE TRIGGER IF NOT EXISTS "%s_member_new_picture_uuid" - AFTER INSERT ON "%s_members" - WHEN NEW.picture_uuid IS NOT NULL - BEGIN - INSERT INTO "%s_member_changes" ( - member_id, attribute, value_blob, op - ) VALUES - (NEW.id, 'picture_uuid', NEW.picture_uuid, true) - ; - END; - CREATE TRIGGER IF NOT EXISTS "%s_member_update_username" - AFTER UPDATE ON "%s_members" - WHEN OLD.username != NEW.username - BEGIN - INSERT INTO "%s_member_changes" ( - member_id, attribute, value_text, op - ) VALUES - (NEW.id, 'username', OLD.username, false), - (NEW.id, 'username', NEW.username, true) - ; - END; - CREATE TRIGGER IF NOT EXISTS "%s_member_update_display_name" - AFTER UPDATE ON "%s_members" - WHEN OLD.display_name != NEW.display_name - BEGIN - INSERT INTO "%s_member_changes" ( - member_id, attribute, value_text, op - ) VALUES - (NEW.id, 'display_name', OLD.display_name, false), - (NEW.id, 'display_name', NEW.display_name, true) - ; - END; - CREATE TRIGGER IF NOT EXISTS "%s_member_update_status" - AFTER UPDATE ON "%s_members" - WHEN OLD.status != NEW.status - BEGIN - INSERT INTO "%s_member_changes" ( - member_id, attribute, value_text, op - ) VALUES - (NEW.id, 'status', OLD.status, false), - (NEW.id, 'status', NEW.status, true) - ; - END; - CREATE TRIGGER IF NOT EXISTS "%s_member_add_picture_uuid" - AFTER UPDATE ON "%s_members" - WHEN ( - OLD.picture_uuid IS NULL AND - NEW.picture_uuid IS NOT NULL - ) - BEGIN - INSERT INTO "%s_member_changes" ( - member_id, attribute, value_blob, op - ) VALUES - (NEW.id, 'picture_uuid', NEW.picture_uuid, true) - ; - END; - CREATE TRIGGER IF NOT EXISTS "%s_member_remove_picture_uuid" - AFTER UPDATE ON "%s_members" - WHEN ( - OLD.picture_uuid IS NOT NULL AND - NEW.picture_uuid IS NULL - ) - BEGIN - INSERT INTO "%s_member_changes" ( - member_id, attribute, value_blob, op - ) VALUES - (NEW.id, 'picture_uuid', OLD.picture_uuid, false) - ; - END; - CREATE TRIGGER IF NOT EXISTS "%s_member_update_picture_uuid" - AFTER UPDATE ON "%s_members" - WHEN ( - OLD.picture_uuid IS NOT NULL AND - NEW.picture_uuid IS NOT NULL AND - OLD.picture_uuid != NEW.picture_uuid - ) - BEGIN - INSERT INTO "%s_member_changes" ( - member_id, attribute, value_blob, op - ) VALUES - (NEW.id, 'picture_uuid', OLD.picture_uuid, false), - (NEW.id, 'picture_uuid', NEW.picture_uuid, true) - ; - END; - - CREATE TABLE IF NOT EXISTS "%s_member_roles" ( - id INTEGER NOT NULL PRIMARY KEY AUTOINCREMENT, - member_id INTEGER NOT NULL - REFERENCES "%s_members"(id), - role TEXT NOT NULL, - UNIQUE (member_id, role) - ) STRICT; - CREATE TABLE IF NOT EXISTS "%s_member_role_changes" ( - id INTEGER NOT NULL PRIMARY KEY AUTOINCREMENT, - timestamp TEXT NOT NULL DEFAULT ( - %s - ), - role_id INTEGER NOT NULL, - attribute TEXT NOT NULL CHECK( - attribute IN ( - 'role', - 'logon_id' -- FIXME - ) - ), - value TEXT NOT NULL, - op INT NOT NULL CHECK(op IN (0, 1)) - ) STRICT; - CREATE TRIGGER IF NOT EXISTS "%s_member_role_add" - AFTER INSERT ON "%s_member_roles" - BEGIN - INSERT INTO "%s_member_role_changes" ( - role_id, attribute, value, op - ) VALUES - (NEW.id, 'role', NEW.role, true) - ; - END; - CREATE TRIGGER IF NOT EXISTS "%s_member_role_remove" - AFTER DELETE ON "%s_member_roles" - BEGIN - INSERT INTO "%s_member_role_changes" ( - role_id, attribute, value, op - ) VALUES - (OLD.id, 'role', OLD.role, false) - ; - END; - CREATE TABLE IF NOT EXISTS "%s_channels" ( - id INTEGER NOT NULL PRIMARY KEY AUTOINCREMENT, - timestamp TEXT NOT NULL DEFAULT ( - %s - ), - uuid BLOB NOT NULL UNIQUE, - network_id INTEGER NOT NULL - REFERENCES "%s_networks"(id), - public_name TEXT, - label TEXT NOT NULL, - description TEXT NOT NULL, - virtual INT NOT NULL CHECK(virtual IN (0, 1)), - UNIQUE (network_id, public_name) - ) STRICT; - CREATE TABLE IF NOT EXISTS "%s_channel_changes" ( - id INTEGER NOT NULL PRIMARY KEY AUTOINCREMENT, - timestamp TEXT NOT NULL DEFAULT ( - %s - ), - channel_id INTEGER NOT NULL, - attribute TEXT NOT NULL CHECK( - attribute IN ( - 'public_name', - 'label', - 'description', - 'virtual', - 'logon_id' -- FIXME - ) - ), - value_text TEXT, - value_bool INT CHECK(value_bool IN (0, 1)), - op INT NOT NULL CHECK(op IN (0, 1)) - ) STRICT; - CREATE TRIGGER IF NOT EXISTS "%s_channel_new" - AFTER INSERT ON "%s_channels" - BEGIN - INSERT INTO "%s_channel_changes" ( - channel_id, attribute, value_text, op - ) VALUES - (NEW.id, 'label', NEW.label, true), - (NEW.id, 'description', NEW.description, true) - ; - INSERT INTO "%s_channel_changes" ( - channel_id, attribute, value_bool, op - ) VALUES - (NEW.id, 'virtual', NEW.virtual, true) - ; - END; - CREATE TRIGGER IF NOT EXISTS "%s_channel_new_public_name" - AFTER INSERT ON "%s_channels" - WHEN NEW.public_name IS NOT NULL - BEGIN - INSERT INTO "%s_channel_changes" ( - channel_id, attribute, value_text, op - ) VALUES - (NEW.id, 'public_name', NEW.public_name, true) - ; - END; - CREATE TRIGGER IF NOT EXISTS "%s_channel_update_label" - AFTER UPDATE ON "%s_channels" - WHEN OLD.label != NEW.label - BEGIN - INSERT INTO "%s_channel_changes" ( - channel_id, attribute, value_text, op - ) VALUES - (NEW.id, 'label', OLD.label, false), - (NEW.id, 'label', NEW.label, true) - ; - END; - CREATE TRIGGER IF NOT EXISTS "%s_channel_update_description" - AFTER UPDATE ON "%s_channels" - WHEN OLD.description != NEW.description - BEGIN - INSERT INTO "%s_channel_changes" ( - channel_id, attribute, value_text, op - ) VALUES - (NEW.id, 'description', OLD.description, false), - (NEW.id, 'description', NEW.description, true) - ; - END; - CREATE TRIGGER IF NOT EXISTS "%s_channel_update_virtual" - AFTER UPDATE ON "%s_channels" - WHEN OLD.virtual != NEW.virtual - BEGIN - INSERT INTO "%s_channel_changes" ( - channel_id, attribute, value_bool, op - ) VALUES - (NEW.id, 'virtual', OLD.virtual, false), - (NEW.id, 'virtual', NEW.virtual, true) - ; - END; - CREATE TRIGGER IF NOT EXISTS "%s_channel_add_public_name" - AFTER UPDATE ON "%s_channels" - WHEN ( - OLD.public_name IS NULL AND - NEW.public_name IS NOT NULL - ) - BEGIN - INSERT INTO "%s_channel_changes" ( - channel_id, attribute, value_text, op - ) VALUES - (NEW.id, 'public_name', NEW.public_name, true) - ; - END; - CREATE TRIGGER IF NOT EXISTS "%s_channel_remove_public_name" - AFTER UPDATE ON "%s_channels" - WHEN ( - OLD.public_name IS NOT NULL AND - NEW.public_name IS NULL - ) - BEGIN - INSERT INTO "%s_channel_changes" ( - channel_id, attribute, value_text, op - ) VALUES - (OLD.id, 'public_name', OLD.public_name, false) - ; - END; - - CREATE TABLE IF NOT EXISTS "%s_participants" ( - id INTEGER NOT NULL PRIMARY KEY AUTOINCREMENT, - channel_id INTEGER NOT NULL - REFERENCES "%s_channels"(id), - member_id INTEGER NOT NULL - REFERENCES "%s_members"(id), - UNIQUE (channel_id, member_id) - ) STRICT; - CREATE TABLE IF NOT EXISTS "%s_participant_changes" ( - id INTEGER NOT NULL PRIMARY KEY AUTOINCREMENT, - timestamp TEXT NOT NULL DEFAULT ( - %s - ), - participant_id INTEGER NOT NULL, - attribute TEXT NOT NULL CHECK( - attribute IN ( - 'connection_id' - ) - ), - value TEXT NOT NULL, - op INT NOT NULL CHECK(op IN (0, 1)) - ) STRICT; - - CREATE TABLE IF NOT EXISTS "%s_channel_events" ( - id INTEGER NOT NULL PRIMARY KEY AUTOINCREMENT, - timestamp TEXT NOT NULL DEFAULT ( - %s - ), - uuid BLOB NOT NULL UNIQUE, - channel_id INTEGER NOT NULL - REFERENCES "%s_channels"(id), - source_uuid BLOB NOT NULL, - source_type TEXT NOT NULL CHECK( - source_type IN ( - 'logon' - ) - ), - source_metadata TEXT, - type TEXT NOT NULL CHECK( - type IN ( - 'user-join', - 'user-message' - ) - ), - payload TEXT NOT NULL, - metadata TEXT - ) STRICT; - - ` - return queryT{ - write: fmt.Sprintf( - tmpl_write, - prefix, - g.SQLiteNow, - prefix, - g.SQLiteNow, - prefix, - prefix, - prefix, - prefix, - prefix, - prefix, - prefix, - prefix, - prefix, - prefix, - prefix, - prefix, - prefix, - prefix, - prefix, - prefix, - prefix, - prefix, - prefix, - prefix, - prefix, - prefix, - prefix, - prefix, - prefix, - prefix, - g.SQLiteNow, - prefix, - prefix, - g.SQLiteNow, - prefix, - g.SQLiteNow, - prefix, - prefix, - prefix, - g.SQLiteNow, - prefix, - prefix, - prefix, - g.SQLiteNow, - prefix, - prefix, - prefix, - prefix, - prefix, - prefix, - prefix, - prefix, - prefix, - prefix, - prefix, - prefix, - prefix, - prefix, - prefix, - prefix, - g.SQLiteNow, - prefix, - prefix, - g.SQLiteNow, - prefix, - prefix, - prefix, - prefix, - prefix, - prefix, - prefix, - prefix, - prefix, - prefix, - prefix, - prefix, - prefix, - prefix, - prefix, - prefix, - prefix, - prefix, - prefix, - prefix, - prefix, - prefix, - prefix, - prefix, - prefix, - prefix, - prefix, - g.SQLiteNow, - prefix, - prefix, - prefix, - prefix, - prefix, - prefix, - prefix, - g.SQLiteNow, - prefix, - prefix, - g.SQLiteNow, - prefix, - prefix, - prefix, - prefix, - prefix, - prefix, - prefix, - prefix, - prefix, - prefix, - prefix, - prefix, - prefix, - prefix, - prefix, - prefix, - prefix, - prefix, - prefix, - prefix, - prefix, - prefix, - prefix, - prefix, - prefix, - prefix, - g.SQLiteNow, - prefix, - g.SQLiteNow, - prefix, - ), - } -} - -func createTables(db *sql.DB, prefix string) error { - q := createTablesSQL(prefix) - - return inTx(db, func(tx *sql.Tx) error { - _, err := tx.Exec(q.write) - return err - }) -} - -func memberRolesSQL(prefix string) queryT { - const tmpl_read = ` - SELECT role FROM "%s_member_roles" - JOIN "%s_members" ON - "%s_member_roles".member_id = "%s_members".id - WHERE "%s_members".uuid = ? - ORDER BY "%s_member_roles".id; - ` - return queryT{ - read: fmt.Sprintf( - tmpl_read, - prefix, - prefix, - prefix, - prefix, - prefix, - prefix, - ), - } -} - -func collectRoles(rows *sql.Rows) ([]string, error) { - roles := []string{} - - for rows.Next() { - var role string - err := rows.Scan(&role) - if err != nil { - rows.Close() - return nil, err - } - - roles = append(roles, role) - } - - return roles, g.WrapErrors(rows.Err(), rows.Close()) -} - -func createUserSQL(prefix string) queryT { - const tmpl_write = ` - INSERT INTO "%s_users" ( - user_uuid, username, display_name, picture_uuid, deleted - ) VALUES ( - ?, ?, ?, NULL, false - ) RETURNING id, timestamp; - ` - return queryT{ - write: fmt.Sprintf(tmpl_write, prefix), - } -} - -func createUserStmt( - cfg dbconfigT, -) (func(newUserT) (userT, error), func() error, error) { - q := createUserSQL(cfg.prefix) - - writeStmt, err := cfg.shared.Prepare(q.write) - if err != nil { - return nil, nil, err - } - - fn := func(newUser newUserT) (userT, error) { - user := userT{ - uuid: newUser.uuid, - username: newUser.username, - displayName: newUser.displayName, - } - - var timestr string - err := writeStmt.QueryRow( - newUser.uuid[:], - newUser.username, - newUser.displayName, - ).Scan(&user.id, ×tr) - if err != nil { - return userT{}, err - } - - user.timestamp, err = time.Parse(time.RFC3339Nano, timestr) - if err != nil { - return userT{}, err - } - - return user, nil - } - - return fn, writeStmt.Close, nil -} - -func userByUUIDSQL(prefix string) queryT { - const tmpl_read = ` - SELECT - id, - timestamp, - username, - display_name, - picture_uuid - FROM "%s_users" - WHERE - user_uuid = ? AND - deleted = false; - ` - return queryT{ - read: fmt.Sprintf(tmpl_read, prefix), - } -} - -func userByUUIDStmt( - cfg dbconfigT, -) (func(uuid.UUID) (userT, error), func() error, error) { - q := userByUUIDSQL(cfg.prefix) - - readStmt, err := cfg.shared.Prepare(q.read) - if err != nil { - return nil, nil, err - } - - fn := func(userID uuid.UUID) (userT, error) { - user := userT{ - uuid: userID, - } - - var ( - timestr string - picture_id_bytes []byte - ) - err := readStmt.QueryRow(userID[:]).Scan( - &user.id, - ×tr, - &user.username, - &user.displayName, - &picture_id_bytes, - ) - if err != nil { - return userT{}, err - } - if picture_id_bytes != nil { - pictureID := uuid.UUID(picture_id_bytes) - user.pictureID = &pictureID - } - - user.timestamp, err = time.Parse(time.RFC3339Nano, timestr) - if err != nil { - return userT{}, err - } - - return user, err - } - - return fn, readStmt.Close, nil -} - -func updateUserSQL(prefix string) queryT { - const tmpl_write = ` - UPDATE "%s_users" - SET - username = ?, - display_name = ?, - picture_uuid = ? - WHERE - id = ? AND - deleted = false - RETURNING id; - ` - return queryT{ - write: fmt.Sprintf(tmpl_write, prefix), - } -} - -func updateUserStmt( - cfg dbconfigT, -) (func(userT) error, func() error, error) { - q := updateUserSQL(cfg.prefix) - - writeStmt, err := cfg.shared.Prepare(q.write) - if err != nil { - return nil, nil, err - } - - fn := func(user userT) error { - var picture_id_bytes []byte = nil - if user.pictureID != nil { - picture_id_bytes = user.pictureID[:] - } - var _id int64 - return writeStmt.QueryRow( - user.username, - user.displayName, - picture_id_bytes, - user.id, - ).Scan(&_id) - } - - return fn, writeStmt.Close, nil -} - -func deleteUserSQL(prefix string) queryT { - const tmpl_write = ` - UPDATE "%s_users" - SET deleted = true - WHERE - user_uuid = ? AND - deleted = false - RETURNING id; - ` - return queryT{ - write: fmt.Sprintf(tmpl_write, prefix), - } -} - -func deleteUserStmt( - cfg dbconfigT, -) (func(uuid.UUID) error, func() error, error) { - q := deleteUserSQL(cfg.prefix) - - writeStmt, err := cfg.shared.Prepare(q.write) - if err != nil { - return nil, nil, err - } - - fn := func(userID uuid.UUID) error { - var _id int64 - return writeStmt.QueryRow(userID[:]).Scan(&_id) - } - - return fn, writeStmt.Close, nil -} - -func addNetworkSQL(prefix string) queryT { - const tmpl_write = ` - INSERT INTO "%s_networks" ( - uuid, name, description, type, deleted - ) - VALUES ( - ?, - ?, - ?, - ?, - false - ) RETURNING id; - - WITH creator AS ( - SELECT username, display_name, picture_uuid - FROM "%s_users" - WHERE id = ? AND deleted = false - ), new_network AS ( - SELECT id FROM "%s_networks" WHERE uuid = ? - ) - INSERT INTO "%s_members" ( - uuid, network_id, user_id, username, display_name, - picture_uuid, status, active_uniq - ) VALUES ( - ?, - (SELECT id FROM new_network), - ?, - (SELECT username FROM creator), - (SELECT display_name FROM creator), - (SELECT picture_uuid FROM creator), - 'active', - 'active' - ) RETURNING id; - - WITH new_member AS ( - SELECT id FROM "%s_members" WHERE uuid = ? - ) - INSERT INTO "%s_member_roles" (member_id, role) - VALUES ( - (SELECT id FROM new_member), - 'admin' - ), - ( - (SELECT id FROM new_member), - 'creator' - ) - RETURNING id; - ` - const tmpl_read = ` - SELECT id, timestamp FROM "%s_networks" - WHERE uuid = ? AND deleted = false; - ` - return queryT{ - write: fmt.Sprintf( - tmpl_write, - prefix, - prefix, - prefix, - prefix, - prefix, - prefix, - ), - read: fmt.Sprintf(tmpl_read, prefix), - } -} - -func addNetworkStmt( - cfg dbconfigT, -) ( - func(userT, newNetworkT, uuid.UUID) (networkT, error), - func() error, - error, -) { - q := addNetworkSQL(cfg.prefix) - - readStmt, err := cfg.shared.Prepare(q.read) - if err != nil { - return nil, nil, err - } - - privateDB, err := sql.Open(golite.DriverName, cfg.dbpath) - if err != nil { - readStmt.Close() - return nil, nil, err - } - - writeFn, writeFnClose := execSerialized(q.write, privateDB) - - fn := func( - user userT, - newNetwork newNetworkT, - memberID uuid.UUID, - ) (networkT, error) { - network := networkT{ - uuid: newNetwork.uuid, - name: newNetwork.name, - description: newNetwork.description, - type_: newNetwork.type_, - } - - err := writeFn( - newNetwork.uuid[:], - newNetwork.name, - newNetwork.description, - newNetwork.type_, - user.id, - newNetwork.uuid[:], - memberID[:], - user.id, - memberID[:], - ) - if err != nil { - return networkT{}, err - } - - var timestr string - err = readStmt.QueryRow(network.uuid[:]).Scan( - &network.id, - ×tr, - ) - if err != nil { - return networkT{}, err - } - - network.timestamp, err = time.Parse(time.RFC3339Nano, timestr) - if err != nil { - return networkT{}, err - } - - return network, nil - } - - closeFn := func() error { - writeFnClose() - return g.SomeError(privateDB.Close(), readStmt.Close()) - } - - return fn, closeFn, nil -} - -func getNetworkSQL(prefix string) queryT { - const tmpl_read = ` - WITH probing_user AS ( - SELECT id FROM "%s_users" - WHERE id = ? AND deleted = false - ), target_network AS ( - SELECT id FROM "%s_networks" - WHERE uuid = ? AND deleted = false - ) - SELECT - id, - timestamp, - name, - description, - type - FROM "%s_networks" - WHERE - uuid = ? AND - deleted = false AND - ? IN probing_user AND - ( - type IN ('public', 'unlisted') OR - ? IN ( - SELECT user_id FROM "%s_members" - WHERE - user_id = ? AND - network_id IN target_network AND - status != 'removed' - ) - ); - ` - return queryT{ - read: fmt.Sprintf( - tmpl_read, - prefix, - prefix, - prefix, - prefix, - prefix, - prefix, - prefix, - prefix, - prefix, - prefix, - prefix, - prefix, - prefix, - prefix, - prefix, - prefix, - prefix, - prefix, - prefix, - prefix, - prefix, - prefix, - ), - } -} - -func getNetworkStmt( - cfg dbconfigT, -) (func(userT, uuid.UUID) (networkT, error), func() error, error) { - q := getNetworkSQL(cfg.prefix) - - readStmt, err := cfg.shared.Prepare(q.read) - if err != nil { - return nil, nil, err - } - - fn := func(user userT, networkID uuid.UUID) (networkT, error) { - network := networkT{ - uuid: networkID, - } - - var timestr string - err := readStmt.QueryRow( - user.id, - networkID[:], - networkID[:], - user.id, - user.id, - user.id, - ).Scan( - &network.id, - ×tr, - &network.name, - &network.description, - &network.type_, - ) - if err != nil { - return networkT{}, err - } - - network.timestamp, err = time.Parse(time.RFC3339Nano, timestr) - if err != nil { - return networkT{}, err - } - - return network, nil - } - - return fn, readStmt.Close, nil -} - -func networkEach(rows *sql.Rows, callback func(networkT) error) error { - if rows == nil { - return nil - } - - for rows.Next() { - var ( - network networkT - timestr string - network_id_bytes []byte - deleted bool - ) - err := rows.Scan( - &network.id, - ×tr, - &network_id_bytes, - &network.name, - &network.description, - &network.type_, - &deleted, - ) - if err != nil { - return g.WrapErrors(rows.Close(), err) - } - network.uuid = uuid.UUID(network_id_bytes) - - if deleted { - return sql.ErrNoRows - } - - network.timestamp, err = time.Parse(time.RFC3339Nano, timestr) - if err != nil { - return g.WrapErrors(rows.Close(), err) - } - - err = callback(network) - if err != nil { - return g.WrapErrors(rows.Close(), err) - } - } - - return g.WrapErrors(rows.Err(), rows.Close()) -} - -func networksSQL(prefix string) queryT { - const tmpl_read = ` - WITH current_user AS ( - SELECT id, deleted FROM "%s_users" WHERE id = ? - ) - SELECT - "%s_networks".id, - "%s_networks".timestamp, - "%s_networks".uuid, - "%s_networks".name, - "%s_networks".description, - "%s_networks".type, - (SELECT deleted FROM current_user) - FROM "%s_networks" - JOIN "%s_members" ON - "%s_networks".id = "%s_members".network_id - WHERE ( - "%s_networks".type = 'public' OR - "%s_networks".id IN ( - SELECT network_id FROM "%s_members" - WHERE user_id IN (SELECT id FROM current_user) - ) - ) AND "%s_networks".deleted = false - ORDER BY "%s_networks".id; - ` - return queryT{ - read: fmt.Sprintf( - tmpl_read, - prefix, - prefix, - prefix, - prefix, - prefix, - prefix, - prefix, - prefix, - prefix, - prefix, - prefix, - prefix, - prefix, - prefix, - prefix, - prefix, - ), - } -} - -func networksStmt( - cfg dbconfigT, -) (func(userT) (*sql.Rows, error), func() error, error) { - q := networksSQL(cfg.prefix) - - readStmt, err := cfg.shared.Prepare(q.read) - if err != nil { - return nil, nil, err - } - - fn := func(user userT) (*sql.Rows, error) { - return readStmt.Query(user.id) - } - - return fn, readStmt.Close, nil -} - -func setNetworkSQL(prefix string) queryT { - const tmpl_write = ` - UPDATE "%s_networks" - SET - name = ?, - description = ?, - type = ? - WHERE id = ? AND deleted = false - RETURNING ( - SELECT CASE WHEN EXISTS ( - SELECT role from "%s_member_roles" - WHERE - member_id = ? AND - role IN ( - 'admin', - 'network-settings-update' - ) AND ? IN ( - SELECT network_id - FROM "%s_members" - WHERE - id = ? AND - status = 'active' - ) - ) THEN true ELSE RAISE( - ABORT, - 'member not allowed to update network data' - ) END - ); - - ` - return queryT{ - write: fmt.Sprintf(tmpl_write, prefix, prefix, prefix), - } -} - -func setNetworkStmt( - cfg dbconfigT, -) (func(memberT, networkT) error, func() error, error) { - q := setNetworkSQL(cfg.prefix) - - writeStmt, err := cfg.shared.Prepare(q.write) - if err != nil { - return nil, nil, err - } - - fn := func(actor memberT, network networkT) error { - var _allowed bool - return writeStmt.QueryRow( - network.name, - network.description, - network.type_, - network.id, - actor.id, - network.id, - actor.id, - ).Scan(&_allowed) - } - - return fn, writeStmt.Close, nil -} - -func nipNetworkSQL(prefix string) queryT { - const tmpl_write = ` - WITH target_network AS ( - SELECT network_id AS id - FROM "%s_members" - WHERE - id = ? AND - status = 'active' - ) - UPDATE "%s_networks" - SET deleted = true - WHERE id IN target_network AND deleted = false - RETURNING ( - SELECT CASE WHEN EXISTS ( - SELECT role FROM "%s_member_roles" - WHERE - member_id = ? AND - role IN ( - 'admin' - ) - ) THEN true ELSE RAISE( - ABORT, - 'member not allowed to delete network' - ) END - ); - ` - return queryT{ - write: fmt.Sprintf(tmpl_write, prefix, prefix, prefix), - } -} - -func nipNetworkStmt( - cfg dbconfigT, -) (func(memberT) error, func() error, error) { - q := nipNetworkSQL(cfg.prefix) - - writeStmt, err := cfg.shared.Prepare(q.write) - if err != nil { - return nil, nil, err - } - - fn := func(actor memberT) error { - var _allowed bool - return writeStmt.QueryRow(actor.id, actor.id).Scan(&_allowed) - } - - return fn, writeStmt.Close, nil -} - -func membershipSQL(prefix string) queryT { - const tmpl_read = ` - SELECT - "%s_members".id, - "%s_members".timestamp, - "%s_members".uuid, - "%s_members".username, - "%s_members".display_name, - "%s_members".picture_uuid, - "%s_members".status - FROM "%s_members" - JOIN "%s_users" ON - "%s_users".id = "%s_members".user_id - JOIN "%s_networks" ON - "%s_networks".id = "%s_members".network_id - WHERE - "%s_members".user_id = ? AND - "%s_members".network_id = ? AND - "%s_members".status = 'active' AND - "%s_users".deleted = false AND - "%s_networks".deleted = false; - ` - return queryT{ - read: fmt.Sprintf( - tmpl_read, - prefix, - prefix, - prefix, - prefix, - prefix, - prefix, - prefix, - prefix, - prefix, - prefix, - prefix, - prefix, - prefix, - prefix, - prefix, - prefix, - prefix, - prefix, - prefix, - ), - } -} - -func membershipStmt( - cfg dbconfigT, -) (func(userT, networkT) (memberT, error), func() error, error) { - q := membershipSQL(cfg.prefix) - - readStmt, err := cfg.shared.Prepare(q.read) - if err != nil { - return nil, nil, err - } - - rolesStmt, err := cfg.shared.Prepare(memberRolesSQL(cfg.prefix).read) - if err != nil { - readStmt.Close() - return nil, nil, err - } - - fn := func(actor userT, network networkT) (memberT, error) { - member := memberT{} - - var ( - timestr string - member_id_bytes []byte - picture_id_bytes []byte - ) - err := readStmt.QueryRow(actor.id, network.id).Scan( - &member.id, - ×tr, - &member_id_bytes, - &member.username, - &member.displayName, - &picture_id_bytes, - &member.status, - ) - if err != nil { - return memberT{}, err - } - member.uuid = uuid.UUID(member_id_bytes) - - member.timestamp, err = time.Parse(time.RFC3339Nano, timestr) - if err != nil { - return memberT{}, err - } - - rows, err := rolesStmt.Query(member_id_bytes) - if err != nil { - return memberT{}, err - } - - member.roles, err = collectRoles(rows) - if err != nil { - return memberT{}, err - } - - return member, nil - } - - closeFn := func() error { - return g.SomeError( - readStmt.Close(), - rolesStmt.Close(), - ) - } - - return fn, closeFn, nil -} - -func addMemberSQL(prefix string) queryT { - const tmpl_write = ` - WITH target_user AS ( - SELECT id, username, display_name, picture_uuid - FROM "%s_users" - WHERE user_uuid = ? AND deleted = false - ), target_network AS ( - SELECT "%s_members".network_id AS id - FROM "%s_members" - JOIN "%s_networks" ON - "%s_members".network_id = "%s_networks".id - WHERE - "%s_members".id = ? AND - "%s_members".status = 'active' AND - "%s_networks".deleted = false - ) - INSERT INTO "%s_members" ( - uuid, network_id, user_id, username, display_name, - picture_uuid, status, active_uniq - ) VALUES ( - ?, - (SELECT id FROM target_network), - (SELECT id FROM target_user), - ?, - (SELECT display_name FROM target_user), - (SELECT picture_uuid FROM target_user), - 'active', - 'active' - ) RETURNING id, timestamp, display_name, picture_uuid, status, ( - SELECT CASE WHEN EXISTS ( - SELECT role from "%s_member_roles" - WHERE - member_id = ? AND - role IN ( - 'admin', - 'add-member' - ) - ) THEN true ELSE RAISE( - ABORT, - 'member not allowed to add another member' - ) END - ); - ` - return queryT{ - write: fmt.Sprintf( - tmpl_write, - prefix, - prefix, - prefix, - prefix, - prefix, - prefix, - prefix, - prefix, - prefix, - prefix, - prefix, - ), - } -} - -func addMemberStmt( - cfg dbconfigT, -) (func(memberT, newMemberT) (memberT, error), func() error, error) { - q := addMemberSQL(cfg.prefix) - - writeStmt, err := cfg.shared.Prepare(q.write) - if err != nil { - return nil, nil, err - } - - rolesStmt, err := cfg.shared.Prepare(memberRolesSQL(cfg.prefix).read) - if err != nil { - writeStmt.Close() - return nil, nil, err - } - - fn := func(actor memberT, newMember newMemberT) (memberT, error) { - member := memberT{ - uuid: newMember.memberID, - username: newMember.username, - } - - var ( - timestr string - picture_id_bytes []byte - _allowed bool - ) - err := writeStmt.QueryRow( - newMember.userID[:], - actor.id, - newMember.memberID[:], - newMember.username, - actor.id, - ).Scan( - &member.id, - ×tr, - &member.displayName, - &picture_id_bytes, - &member.status, - &_allowed, - ) - if err != nil { - return memberT{}, err - } - if picture_id_bytes != nil { - pictureID := uuid.UUID(picture_id_bytes) - member.pictureID = &pictureID - } - - member.timestamp, err = time.Parse(time.RFC3339Nano, timestr) - if err != nil { - return memberT{}, err - } - - rows, err := rolesStmt.Query(member.uuid[:]) - if err != nil { - return memberT{}, err - } - - member.roles, err = collectRoles(rows) - if err != nil { - return memberT{}, err - } - - return member, nil - } - - closeFn := func() error { - return g.SomeError( - writeStmt.Close(), - rolesStmt.Close(), - ) - } - - return fn, closeFn, nil -} - -func addRoleSQL(prefix string) queryT { - const tmpl_write = ` - INSERT INTO "%s_member_roles" (member_id, role) - VALUES (?, ?); - ` - return queryT{ - write: fmt.Sprintf(tmpl_write, prefix), - } -} - -func addRoleStmt( - cfg dbconfigT, -) (func(memberT, string, memberT) error, func() error, error) { - q := addRoleSQL(cfg.prefix) - - writeStmt, err := cfg.shared.Prepare(q.write) - if err != nil { - return nil, nil, err - } - - fn := func(actor memberT, role string, member memberT) error { - // FIXME: do authorization - _, err := writeStmt.Exec(member.id, role) - return err - } - - return fn, writeStmt.Close, nil -} - -func dropRoleSQL(prefix string) queryT { - const tmpl_write = ` - DELETE FROM "%s_member_roles" - WHERE - member_id = ? AND - role = ? - RETURNING 1; - ` - return queryT{ - write: fmt.Sprintf(tmpl_write, prefix), - } -} - -func dropRoleStmt( - cfg dbconfigT, -) (func(memberT, string, memberT) error, func() error, error) { - q := dropRoleSQL(cfg.prefix) - - writeStmt, err := cfg.shared.Prepare(q.write) - if err != nil { - return nil, nil, err - } - - fn := func(actor memberT, role string, member memberT) error { - // FIXME: do authorization - // _, err := writeStmt.Exec(member.id, role) - // return err - var _id int64 - return writeStmt.QueryRow(member.id, role).Scan(&_id) - } - - return fn, writeStmt.Close, nil -} - - -func showMemberSQL(prefix string) queryT { - const tmpl_read = ` - WITH current_network AS ( - SELECT network_id - FROM "%s_members" - WHERE id = ? - ) - SELECT - id, - timestamp, - username, - display_name, - picture_uuid, - status - FROM "%s_members" - WHERE - uuid = ? AND - network_id IN current_network; - ` - return queryT{ - read: fmt.Sprintf(tmpl_read, prefix, prefix), - } -} - -func showMemberStmt( - cfg dbconfigT, -) (func(memberT, uuid.UUID) (memberT, error), func() error, error) { - q := showMemberSQL(cfg.prefix) - - readStmt, err := cfg.shared.Prepare(q.read) - if err != nil { - return nil, nil, err - } - - rolesStmt, err := cfg.shared.Prepare(memberRolesSQL(cfg.prefix).read) - if err != nil { - readStmt.Close() - return nil, nil, err - } - - fn := func(actor memberT, memberID uuid.UUID) (memberT, error) { - member := memberT{ - uuid: memberID, - } - - var ( - timestr string - picture_id_bytes []byte - ) - err := readStmt.QueryRow(actor.id, memberID[:]).Scan( - &member.id, - ×tr, - &member.username, - &member.displayName, - &picture_id_bytes, - &member.status, - ) - if err != nil { - return memberT{}, err - } - if picture_id_bytes != nil { - pictureID := uuid.UUID(picture_id_bytes) - // FIXME: test this - member.pictureID = &pictureID - } - - member.timestamp, err = time.Parse(time.RFC3339Nano, timestr) - if err != nil { - return memberT{}, err - } - - rows, err := rolesStmt.Query(memberID[:]) - if err != nil { - return memberT{}, err - } - - member.roles, err = collectRoles(rows) - if err != nil { - return memberT{}, err - } - - return member, nil - } - - closeFn := func() error { - return g.SomeError( - readStmt.Close(), - rolesStmt.Close(), - ) - } - - return fn, closeFn, nil -} - -func memberEach(rows *sql.Rows, callback func(memberT) error) error { - if rows == nil { - return nil - } - - for rows.Next() { - var ( - member memberT - timestr string - member_id_bytes []byte - picture_id_bytes []byte - ) - err := rows.Scan( - &member.id, - ×tr, - &member_id_bytes, - &member.username, - &member.displayName, - &picture_id_bytes, - &member.status, - ) - if err != nil { - return g.WrapErrors(rows.Close(), err) - } - member.uuid = uuid.UUID(member_id_bytes) - if picture_id_bytes != nil { - pictureID := uuid.UUID(picture_id_bytes) - member.pictureID = &pictureID - } - - member.timestamp, err = time.Parse(time.RFC3339Nano, timestr) - if err != nil { - return g.WrapErrors(rows.Close(), err) - } - - err = callback(member) - if err != nil { - return g.WrapErrors(rows.Close(), err) - } - } - - return g.WrapErrors(rows.Err(), rows.Close()) -} - -func membersSQL(prefix string) queryT { - const tmpl_read = ` - WITH target_network AS ( - SELECT "%s_members".network_id - FROM "%s_members" - JOIN "%s_networks" ON - "%s_members".network_id = "%s_networks".id - WHERE - "%s_members".id = ? AND - "%s_networks".deleted = false - ) - SELECT - id, - timestamp, - uuid, - username, - display_name, - picture_uuid, - status - FROM "%s_members" - WHERE - network_id IN target_network AND - status = 'active'; - ` - return queryT{ - read: fmt.Sprintf( - tmpl_read, - prefix, - prefix, - prefix, - prefix, - prefix, - prefix, - prefix, - prefix, - ), - } -} - -func membersStmt( - cfg dbconfigT, -) (func(memberT) (*sql.Rows, error), func() error, error) { - q := membersSQL(cfg.prefix) - - readStmt, err := cfg.shared.Prepare(q.read) - if err != nil { - return nil, nil, err - } - - fn := func(actor memberT) (*sql.Rows, error) { - return readStmt.Query(actor.id) - } - - return fn, readStmt.Close, nil -} - -func editMemberSQL(prefix string) queryT { - const tmpl_write = ` - UPDATE "%s_members" - SET - status = ? - WHERE id = ? - RETURNING id; - ` - return queryT{ - write: fmt.Sprintf(tmpl_write, prefix), - } -} - -func editMemberStmt( - cfg dbconfigT, -) (func(memberT, memberT) error, func() error, error) { - q := editMemberSQL(cfg.prefix) - - writeStmt, err := cfg.shared.Prepare(q.write) - if err != nil { - return nil, nil, err - } - - fn := func(actor memberT, member memberT) error { - var _id int64 - return writeStmt.QueryRow( - member.status, - member.id, - ).Scan(&_id) - } - - return fn, writeStmt.Close, nil -} - -func dropMemberSQL(prefix string) queryT { - const tmpl_write = ` - UPDATE "%s_members" SET status = 'removed' - WHERE uuid = ? RETURNING id; - - DELETE FROM "%s_member_roles" - WHERE - role != 'creator' AND - member_id IN ( - SELECT id FROM "%s_members" - WHERE uuid = ? - ) - ` - return queryT{ - write: fmt.Sprintf(tmpl_write, prefix, prefix, prefix), - } -} - -func dropMemberStmt( - cfg dbconfigT, -) (func(memberT, uuid.UUID) error, func() error, error) { - q := dropMemberSQL(cfg.prefix) - - privateDB, err := sql.Open(golite.DriverName, cfg.dbpath) - if err != nil { - return nil, nil, err - } - - writeFn, writeFnClose := execSerialized(q.write, privateDB) - - fn := func(actor memberT, memberID uuid.UUID) error { - err := writeFn(memberID[:], memberID[:]) - if err != nil { - return err - } - - // if res == 0 { // FIXME } - return nil - } - - closeFn := func() error { - writeFnClose() - return privateDB.Close() - } - - return fn, closeFn, nil -} - -func addChannelSQL(prefix string) queryT { - const tmpl_write = ` - WITH target_network AS ( - SELECT network_id AS id - FROM "%s_members" - WHERE id = ? - ) - INSERT INTO "%s_channels" ( - uuid, - network_id, - public_name, - label, - description, - virtual - ) VALUES ( - ?, - (SELECT id FROM target_network), - ?, - ?, - ?, - ? - ) RETURNING id, timestamp; - - WITH new_channel AS ( - SELECT id FROM "%s_channels" WHERE uuid = ? - ) - INSERT INTO "%s_participants" (channel_id, member_id) - VALUES ( - (SELECT id FROM new_channel), - ? - ); - ` - const tmpl_read = ` - SELECT id, timestamp FROM "%s_channels" - WHERE uuid = ?; - ` - return queryT{ - write: fmt.Sprintf(tmpl_write, prefix, prefix, prefix, prefix), - read: fmt.Sprintf(tmpl_read, prefix), - } -} - -func addChannelStmt( - cfg dbconfigT, -) (func (memberT, newChannelT) (channelT, error), func() error, error) { - q := addChannelSQL(cfg.prefix) - - readStmt, err := cfg.shared.Prepare(q.read) - if err != nil { - return nil, nil, err - } - - privateDB, err := sql.Open(golite.DriverName, cfg.dbpath) - if err != nil { - readStmt.Close() - return nil, nil, err - } - - writeFn, writeFnClose := execSerialized(q.write, privateDB) - - fn := func(actor memberT, newChannel newChannelT) (channelT, error) { - channel := channelT{ - uuid: newChannel.uuid, - publicName: newChannel.publicName, - label: newChannel.label, - description: newChannel.description, - virtual: newChannel.virtual, - } - - var timestr string - err := writeFn( - actor.id, - newChannel.uuid[:], - newChannel.publicName, - newChannel.label, - newChannel.description, - newChannel.virtual, - newChannel.uuid[:], - actor.id, - ) - if err != nil { - return channelT{}, err - } - - err = readStmt.QueryRow(newChannel.uuid[:]).Scan( - &channel.id, - ×tr, - ) - if err != nil { - return channelT{}, err - } - - channel.timestamp, err = time.Parse(time.RFC3339Nano, timestr) - if err != nil { - return channelT{}, err - } - - return channel, nil - } - - closeFn := func() error { - writeFnClose() - return readStmt.Close() - } - - return fn, closeFn, nil -} - -/* -func chanByName(prefix string) queryT { - const tmpl_read = ` - SELECT - id, - timestamp, - uuid, - public, - - ` - return queryT{ - read: fmt.Sprintf(tmpl_read, prefix), - } -} - -func chanByStmt( - cfg dbconfigT, -) ( -FIXME -*/ - -func channelEach(rows *sql.Rows, callback func(channelT) error) error { - if rows == nil { - return nil - } - - for rows.Next() { - var ( - channel channelT - timestr string - channel_id_bytes []byte - publicName sql.NullString - ) - err := rows.Scan( - &channel.id, - ×tr, - &channel_id_bytes, - &publicName, - &channel.label, - &channel.description, - &channel.virtual, - ) - if err != nil { - return g.WrapErrors(rows.Close(), err) - } - channel.uuid = uuid.UUID(channel_id_bytes) - if publicName.Valid { - channel.publicName = &publicName.String - } - - channel.timestamp, err = time.Parse(time.RFC3339Nano, timestr) - if err != nil { - return g.WrapErrors(rows.Close(), err) - } - - err = callback(channel) - if err != nil { - return g.WrapErrors(rows.Close(), err) - } - } - - return g.WrapErrors(rows.Err(), rows.Close()) -} - -func channelsSQL(prefix string) queryT { - const tmpl_read = ` - WITH current_network AS ( - SELECT network_id AS id - FROM "%s_members" - WHERE id = ? - ), member_private_channels AS ( - SELECT channel_id AS id - FROM "%s_participants" - WHERE member_id = ? - ) - SELECT - id, - timestamp, - uuid, - public_name, - label, - description, - virtual - FROM "%s_channels" - WHERE - network_id IN current_network AND - ( - public_name IS NOT NULL OR - id IN member_private_channels - ) - ORDER BY id; - ` - return queryT{ - read: fmt.Sprintf(tmpl_read, prefix, prefix, prefix), - } -} - -func channelsStmt( - cfg dbconfigT, -) (func(memberT) (*sql.Rows, error), func() error, error) { - q := channelsSQL(cfg.prefix) - - readStmt, err := cfg.shared.Prepare(q.read) - if err != nil { - return nil, nil, err - } - - fn := func(actor memberT) (*sql.Rows, error) { - return readStmt.Query(actor.id, actor.id) - } - - return fn, readStmt.Close, nil -} - -func setChannelSQL(prefix string) queryT { - const tmpl_write = ` - WITH participant_channel AS ( - SELECT channel_id AS id - FROM "%s_participants" - WHERE - member_id = ? AND - channel_id = ? - ) - UPDATE "%s_channels" - SET - description = ?, - public_name = ? - WHERE id IN participant_channel - RETURNING id; - ` - const tmpl_read = ` - SELECT ( - SELECT network_id AS id - FROM "%s_channels" - WHERE id = ? - ) AS channel_network_id, ( - SELECT network_id AS id - FROM "%s_members" - WHERE id = ? - ) AS member_network_id; - ` - return queryT{ - write: fmt.Sprintf(tmpl_write, prefix, prefix), - read: fmt.Sprintf(tmpl_read, prefix, prefix), - } -} - -func setChannelStmt( - cfg dbconfigT, -) (func(memberT, channelT) error, func() error, error) { - q := setChannelSQL(cfg.prefix) - - readStmt, err := cfg.shared.Prepare(q.read) - if err != nil { - return nil, nil, err - } - - writeStmt, err := cfg.shared.Prepare(q.write) - if err != nil { - readStmt.Close() - return nil, nil, err - } - - fn := func(actor memberT, channel channelT) error { - var ( - netid1 sql.NullInt64 - netid2 sql.NullInt64 - ) - err := readStmt.QueryRow(channel.id, actor.id).Scan( - &netid1, - &netid2, - ) - if err != nil { - return err - } - if !netid1.Valid || !netid2.Valid || - netid1.Int64 != netid2.Int64 { - return sql.ErrNoRows - } - - var _id int64 - return writeStmt.QueryRow( - actor.id, - channel.id, - channel.description, - channel.publicName, - ).Scan(&_id) - } - - closeFn := func() error { - return g.SomeError( - readStmt.Close(), - writeStmt.Close(), - ) - } - - return fn, closeFn, nil -} - -func endChannelSQL(prefix string) queryT { - const tmpl_write = ` - -- FIXME %s - ` - return queryT{ - write: fmt.Sprintf(tmpl_write, prefix), - } -} - -func endChannelStmt( - cfg dbconfigT, -) (func(memberT, uuid.UUID) error, func() error, error) { - q := endChannelSQL(cfg.prefix) - - writeStmt, err := cfg.shared.Prepare(q.write) - if err != nil { - return nil, nil, err - } - - fn := func(actor memberT, channelID uuid.UUID) error { - _, err := writeStmt.Exec(channelID[:]) - return err - } - - return fn, writeStmt.Close, nil -} - -func joinSQL(prefix string) queryT { - const tmpl_write = ` - WITH target_channel AS ( - SELECT id - FROM "%s_channels" - WHERE - uuid = ? AND - public_name IS NOT NULL - ) - INSERT INTO "%s_participants" (channel_id, member_id) - VALUES ( - (SELECT id FROM target_channel), - ? - ) RETURNING id; - ` - const tmpl_read = ` - SELECT ( - SELECT network_id AS id - FROM "%s_channels" - WHERE - uuid = ? AND - public_name IS NOT NULL - ) AS channel_network_id, ( - SELECT network_id AS id - FROM "%s_members" WHERE id = ? - ) AS member_network_id; - ` - return queryT{ - write: fmt.Sprintf(tmpl_write, prefix, prefix), - read: fmt.Sprintf(tmpl_read, prefix, prefix), - } -} - -func joinStmt( - cfg dbconfigT, -) (func(memberT, uuid.UUID) error, func() error, error) { - q := joinSQL(cfg.prefix) - - readStmt, err := cfg.shared.Prepare(q.read) - if err != nil { - return nil, nil, err - } - - writeStmt, err := cfg.shared.Prepare(q.write) - if err != nil { - readStmt.Close() - return nil, nil, err - } - - fn := func(actor memberT, channelID uuid.UUID) error { - var ( - netid1 sql.NullInt64 - netid2 sql.NullInt64 - ) - err := readStmt.QueryRow(channelID[:], actor.id).Scan( - &netid1, - &netid2, - ) - if err != nil { - return err - } - - if !netid1.Valid || !netid2.Valid || - netid1.Int64 != netid2.Int64 { - return sql.ErrNoRows - } - - var _id int64 - return writeStmt.QueryRow(channelID[:], actor.id).Scan(&_id) - } - - closeFn := func() error { - return g.SomeError( - readStmt.Close(), - writeStmt.Close(), - ) - } - - return fn, closeFn, nil -} - -func partSQL(prefix string) queryT { - const tmpl_write = ` - WITH target_channel AS ( - SELECT id - FROM "%s_channels" - WHERE - id = ? AND - virtual = false - ) - DELETE FROM "%s_participants" - WHERE - member_id = ? AND - channel_id IN target_channel - RETURNING 1; - ` - return queryT{ - write: fmt.Sprintf(tmpl_write, prefix, prefix), - } -} - -func partStmt( - cfg dbconfigT, -) (func(memberT, channelT) error, func() error, error) { - q := partSQL(cfg.prefix) - - writeStmt, err := cfg.shared.Prepare(q.write) - if err != nil { - return nil, nil, err - } - - fn := func(actor memberT, channel channelT) error { - var _id int64 - return writeStmt.QueryRow(channel.id, actor.id).Scan(&_id) - } - - return fn, writeStmt.Close, nil -} - -func nameEach(rows *sql.Rows, callback func(memberT) error) error { - if rows == nil { - return nil - } - - for rows.Next() { - var ( - member memberT - timestr string - member_id_bytes []byte - ) - err := rows.Scan( - &member.id, - ×tr, - &member_id_bytes, - ) - if err != nil { - return g.WrapErrors(rows.Close(), err) - } - member.uuid = uuid.UUID(member_id_bytes) - - member.timestamp, err = time.Parse(time.RFC3339Nano, timestr) - if err != nil { - return g.WrapErrors(rows.Close(), err) - } - - err = callback(member) - if err != nil { - return g.WrapErrors(rows.Close(), err) - } - } - - return g.WrapErrors(rows.Err(), rows.Close()) -} - -func namesSQL(prefix string) queryT { - const tmpl_read = ` - -- FIXME %s - ` - return queryT{ - read: fmt.Sprintf(tmpl_read, prefix), - } -} - -func namesStmt( - cfg dbconfigT, -) (func(memberT, uuid.UUID) (*sql.Rows, error), func() error, error) { - q := namesSQL(cfg.prefix) - - readStmt, err := cfg.shared.Prepare(q.read) - if err != nil { - return nil, nil, err - } - - fn := func(actor memberT, channelID uuid.UUID) (*sql.Rows, error) { - return readStmt.Query(channelID[:]) - } - - return fn, readStmt.Close, nil -} - -func addEventSQL(prefix string) queryT { - const tmpl_write = ` - INSERT INTO "%s_channel_events" ( - uuid, channel_id, source_uuid, source_type, - source_metadata, type, payload, metadata - ) VALUES ( - ?, - (SELECT id FROM "%s_channels" WHERE uuid = ?), - ?, - ?, - ?, - ?, - ?, - ? - ) RETURNING id, timestamp; - ` - return queryT{ - write: fmt.Sprintf(tmpl_write, prefix, prefix), - } -} - -func addEventStmt( - cfg dbconfigT, -) (func (newEventT) (eventT, error), func() error, error) { - q := addEventSQL(cfg.prefix) - - writeStmt, err := cfg.shared.Prepare(q.write) - if err != nil { - return nil, nil, err - } - - fn := func(newEvent newEventT) (eventT, error) { - event := eventT{ - uuid: newEvent.eventID, - channelID: newEvent.channelID, - source: newEvent.source, - type_: newEvent.type_, - payload: newEvent.payload, - metadata: newEvent.metadata, - } - - var timestr string - err := writeStmt.QueryRow( - newEvent.eventID[:], - newEvent.channelID[:], - newEvent.source.uuid[:], - newEvent.source.type_, - newEvent.source.metadata, - newEvent.type_, - newEvent.payload, - newEvent.metadata, - ).Scan(&event.id, ×tr) - if err != nil { - return eventT{}, err - } - - event.timestamp, err = time.Parse(time.RFC3339Nano, timestr) - if err != nil { - return eventT{}, err - } - - return event, nil - } - - return fn, writeStmt.Close, nil -} - -func eventEach(rows *sql.Rows, callback func(eventT) error) error { - if rows == nil { - return nil - } - - for rows.Next() { - var ( - event eventT - timestr string - event_id_bytes []byte - channel_id_bytes []byte - ) - err := rows.Scan( - &event.id, - ×tr, - &event_id_bytes, - &channel_id_bytes, - &event.type_, - &event.payload, - ) - if err != nil { - rows.Close() - return err - } - event.uuid = uuid.UUID(event_id_bytes) - event.channelID = uuid.UUID(channel_id_bytes) - - event.timestamp, err = time.Parse(time.RFC3339Nano, timestr) - if err != nil { - rows.Close() - return err - } - - err = callback(event) - if err != nil { - rows.Close() - return err - } - } - - return g.WrapErrors(rows.Err(), rows.Close()) -} - -func allAfterSQL(prefix string) queryT { - const tmpl_read = ` - WITH landmark_event AS ( - SELECT id, channel_id - FROM "%s_channel_events" - WHERE uuid = ? - ) - SELECT - "%s_channel_events".id, - "%s_channel_events".timestamp, - "%s_channel_events".uuid, - "%s_channels".uuid, - -- "%s_channel_events".connection_uuid, - "%s_channel_events".type, - "%s_channel_events".payload - FROM "%s_channel_events" - JOIN "%s_channels" ON - "%s_channel_events".channel_id = "%s_channels".id - WHERE - "%s_channel_events".id > ( - SELECT id FROM landmark_event - ) AND channel_id = ( - SELECT channel_id FROM landmark_event - ); - ` - return queryT{ - read: fmt.Sprintf( - tmpl_read, - prefix, - prefix, - prefix, - prefix, - prefix, - prefix, - prefix, - prefix, - prefix, - prefix, - prefix, - prefix, - prefix, - ), - } -} - -func allAfterStmt( - cfg dbconfigT, -) (func (memberT, uuid.UUID) (*sql.Rows, error), func() error, error) { - q := allAfterSQL(cfg.prefix) - - readStmt, err := cfg.shared.Prepare(q.read) - if err != nil { - return nil, nil, err - } - - fn := func(actor memberT, eventID uuid.UUID) (*sql.Rows, error) { - return readStmt.Query(eventID[:]) - } - - return fn, readStmt.Close, nil -} - -func logMessageSQL(prefix string) queryT{ - const tmpl_write = ` - -- FIXME %s - ` - return queryT{ - write: fmt.Sprintf(tmpl_write, prefix), - } -} - -func logMessageStmt( - cfg dbconfigT, -) (func(userT, messageT) error, func() error, error) { - q := logMessageSQL(cfg.prefix) - - writeStmt, err := cfg.shared.Prepare(q.write) - if err != nil { - return nil, nil, err - } - - // FIXME: actor? - fn := func(user userT, message messageT) error { - return nil // FIXME - _, err := writeStmt.Exec(user, message) - return err - } - - return fn, writeStmt.Close, nil -} - -func initDB( - dbpath string, - prefix string, -) (queriesT, error) { - err := g.ValidateSQLTablePrefix(prefix) - if err != nil { - return queriesT{}, err - } - - shared, err := sql.Open(golite.DriverName, dbpath) - if err != nil { - return queriesT{}, err - } - - cfg := dbconfigT{ - shared: shared, - dbpath: dbpath, - prefix: prefix, - } - - createTablesErr := createTables(shared, prefix) - createUser, createUserClose, createUserErr := createUserStmt(cfg) - userByUUID, userByUUIDClose, userByUUIDErr := userByUUIDStmt(cfg) - updateUser, updateUserClose, updateUserErr := updateUserStmt(cfg) - deleteUser, deleteUserClose, deleteUserErr := deleteUserStmt(cfg) - addNetwork, addNetworkClose, addNetworkErr := addNetworkStmt(cfg) - getNetwork, getNetworkClose, getNetworkErr := getNetworkStmt(cfg) - networks, networksClose, networksErr := networksStmt(cfg) - setNetwork, setNetworkClose, setNetworkErr := setNetworkStmt(cfg) - nipNetwork, nipNetworkClose, nipNetworkErr := nipNetworkStmt(cfg) - membership, membershipClose, membershipErr := membershipStmt(cfg) - addMember, addMemberClose, addMemberErr := addMemberStmt(cfg) - addRole, addRoleClose, addRoleErr := addRoleStmt(cfg) - dropRole, dropRoleClose, dropRoleErr := dropRoleStmt(cfg) - showMember, showMemberClose, showMemberErr := showMemberStmt(cfg) - members, membersClose, membersErr := membersStmt(cfg) - editMember, editMemberClose, editMemberErr := editMemberStmt(cfg) - dropMember, dropMemberClose, dropMemberErr := dropMemberStmt(cfg) - addChannel, addChannelClose, addChannelErr := addChannelStmt(cfg) - channels, channelsClose, channelsErr := channelsStmt(cfg) - setChannel, setChannelClose, setChannelErr := setChannelStmt(cfg) - endChannel, endChannelClose, endChannelErr := endChannelStmt(cfg) - join, joinClose, joinErr := joinStmt(cfg) - part, partClose, partErr := partStmt(cfg) - names, namesClose, namesErr := namesStmt(cfg) - addEvent, addEventClose, addEventErr := addEventStmt(cfg) - allAfter, allAfterClose, allAfterErr := allAfterStmt(cfg) - logMessage, logMessageClose, logMessageErr := logMessageStmt(cfg) - - closeFn := func() error { - return g.SomeFnError( - createUserClose, - userByUUIDClose, - updateUserClose, - deleteUserClose, - addNetworkClose, - getNetworkClose, - networksClose, - setNetworkClose, - nipNetworkClose, - membershipClose, - addMemberClose, - addRoleClose, - dropRoleClose, - showMemberClose, - membersClose, - editMemberClose, - dropMemberClose, - addChannelClose, - channelsClose, - setChannelClose, - endChannelClose, - joinClose, - partClose, - namesClose, - addEventClose, - allAfterClose, - logMessageClose, - ) - } - - err = g.SomeError( - createTablesErr, - createUserErr, - userByUUIDErr, - updateUserErr, - deleteUserErr, - addNetworkErr, - getNetworkErr, - networksErr, - setNetworkErr, - nipNetworkErr, - membershipErr, - addMemberErr, - addRoleErr, - dropRoleErr, - showMemberErr, - membersErr, - editMemberErr, - dropMemberErr, - addChannelErr, - channelsErr, - setChannelErr, - endChannelErr, - joinErr, - partErr, - namesErr, - addEventErr, - allAfterErr, - logMessageErr, - ) - if err != nil { - closeFn() - return queriesT{}, err - } - - var connMutex sync.RWMutex - return queriesT{ - createUser: func(a newUserT) (userT, error) { - connMutex.RLock() - defer connMutex.RUnlock() - return createUser(a) - }, - userByUUID: func(a uuid.UUID) (userT, error) { - connMutex.RLock() - defer connMutex.RUnlock() - return userByUUID(a) - }, - updateUser: func(a userT) error { - connMutex.RLock() - defer connMutex.RUnlock() - return updateUser(a) - }, - deleteUser: func(a uuid.UUID) error { - connMutex.RLock() - defer connMutex.RUnlock() - return deleteUser(a) - }, - addNetwork: func( - a userT, - b newNetworkT, - c uuid.UUID, - ) (networkT, error) { - connMutex.RLock() - defer connMutex.RUnlock() - return addNetwork(a, b, c) - }, - getNetwork: func(a userT, b uuid.UUID) (networkT, error) { - connMutex.RLock() - defer connMutex.RUnlock() - return getNetwork(a, b) - }, - networks: func( - a userT, - callback func(networkT) error, - ) error { - var ( - err error - rows *sql.Rows - ) - { - connMutex.RLock() - defer connMutex.RUnlock() - rows, err = networks(a) - } - if err != nil { - return err - } - - return networkEach(rows, callback) - }, - setNetwork: func(a memberT, b networkT) error { - connMutex.RLock() - defer connMutex.RUnlock() - return setNetwork(a, b) - }, - nipNetwork: func(a memberT) error { - connMutex.RLock() - defer connMutex.RUnlock() - return nipNetwork(a) - }, - membership: func(a userT, b networkT) (memberT, error) { - connMutex.RLock() - defer connMutex.RUnlock() - return membership(a, b) - }, - addMember: func(a memberT, b newMemberT) (memberT, error) { - connMutex.RLock() - defer connMutex.RUnlock() - return addMember(a, b) - }, - addRole: func(a memberT, b string, c memberT) error { - connMutex.RLock() - defer connMutex.RUnlock() - return addRole(a, b, c) - }, - dropRole: func(a memberT, b string, c memberT) error { - connMutex.RLock() - defer connMutex.RUnlock() - return dropRole(a, b, c) - }, - showMember: func(a memberT, b uuid.UUID) (memberT, error) { - connMutex.RLock() - defer connMutex.RUnlock() - return showMember(a, b) - }, - members: func(a memberT, callback func(memberT) error) error { - var ( - err error - rows *sql.Rows - ) - { - connMutex.RLock() - defer connMutex.RUnlock() - rows, err = members(a) - } - if err != nil { - return err - } - - return memberEach(rows, callback) - }, - editMember: func(a memberT, b memberT) error { - connMutex.RLock() - defer connMutex.RUnlock() - return editMember(a, b) - }, - dropMember: func(a memberT, b uuid.UUID) error { - connMutex.RLock() - defer connMutex.RUnlock() - return dropMember(a, b) - }, - addChannel: func( - a memberT, b newChannelT, - ) (channelT, error) { - connMutex.RLock() - defer connMutex.RUnlock() - return addChannel(a, b) - }, - channels: func( - a memberT, - callback func(channelT) error, - ) error { - var ( - err error - rows *sql.Rows - ) - { - connMutex.RLock() - defer connMutex.RUnlock() - rows, err = channels(a) - } - if err != nil { - return err - } - - return channelEach(rows, callback) - }, - setChannel: func(a memberT, b channelT) error { - connMutex.RLock() - defer connMutex.RUnlock() - return setChannel(a, b) - }, - endChannel: func(a memberT, b uuid.UUID) error { - connMutex.RLock() - defer connMutex.RUnlock() - return endChannel(a, b) - }, - join: func(a memberT, b uuid.UUID) error { - connMutex.RLock() - defer connMutex.RUnlock() - return join(a, b) - }, - part: func(a memberT, b channelT) error { - connMutex.RLock() - defer connMutex.RUnlock() - return part(a, b) - }, - names: func( - a memberT, - b uuid.UUID, - callback func(memberT) error, - ) error { - var ( - err error - rows *sql.Rows - ) - { - connMutex.RLock() - defer connMutex.RUnlock() - rows, err = names(a, b) - } - if err != nil { - return err - } - - return nameEach(rows, callback) - }, - addEvent: func(a newEventT) (eventT, error) { - connMutex.RLock() - defer connMutex.RUnlock() - return addEvent(a) - }, - allAfter: func( - a memberT, - b uuid.UUID, - callback func(eventT) error, - ) error { - var ( - err error - rows *sql.Rows - ) - { - connMutex.RLock() - defer connMutex.RUnlock() - rows, err = allAfter(a, b) - } - if err != nil { - return err - } - - return eventEach(rows, callback) - }, - logMessage: func(a userT, b messageT) error { - connMutex.Lock() - defer connMutex.Unlock() - return logMessage(a, b) - }, - close: func() error { - connMutex.Lock() - defer connMutex.Unlock() - return closeFn() - }, - }, nil -} - -func newChannelHandler(papod papodT) func(fiinha.Message) error { - return func(message fiinha.Message) error { - return nil - } -} - -func buildConsumers(prefix string) []consumerT { - return []consumerT{ - consumerT{ - topic: NEW_CHANNEL, - name: prefix + NEW_CHANNEL, - handlerFn: newChannelHandler, - }, - } -} - -func unregisterConsumers( - unsubscribeFn func(string, string), - consumers []consumerT, -) { - for _, c := range consumers { - unsubscribeFn(c.topic, c.name) - } -} - -func registerConsumers( - subscribeFn func(string, string, func(fiinha.Message) error) error, - unsubscribeFn func(string, string), - papod papodT, - consumers []consumerT, -) error { - for _, c := range consumers { - err := subscribeFn(c.topic, c.name, c.handlerFn(papod)) - if err != nil { - unregisterConsumers(unsubscribeFn, consumers) - return err - } - } - return nil -} func initListeners( daemonSocketPath string, commanderSocketPath string, ) (listenersT, error) { + _ = os.Remove(daemonSocketPath) + _ = os.Remove(commanderSocketPath) + daemon, err := net.Listen("unix", daemonSocketPath) if err != nil { return listenersT{}, err @@ -3525,202 +145,89 @@ func initListeners( daemon: daemon, commander: commander, close: func() error { - return g.SomeError( - daemon.Close(), - commander.Close(), + return g.SomeFnError( + daemon.Close, + commander.Close, ) }, }, nil } -// TODO: lock is global, should be by network -func newStateMutable() stateMutableT { - var rwmutex sync.RWMutex - state := stateMutableDataT{ - connections: map[uuid.UUID]*connectionT{}, - users: map[string][]uuid.UUID{}, - members: map[string]map[string][]uuid.UUID{}, - } - return stateMutableT{ - /* - connected: func(connection *connectionT) { - rwmutex.Lock() - defer rwmutex.Unlock() - state.connections[connection.uuid] = connection - }, - disconnect: func(connection *connectionT) { - { - rwmutex.Lock() - defer rwmutex.Unlock() - delete(state.connections, connection.uuid) - delete(state.users, connection.user.username) - delete(state.members, connection.user.username) - } - err := connection.conn.Close() - if err != nil { - g.Warning( - "Failed to close the connection", - "close-error", - "from", "daemon", - "err", err, - ) - } - }, - authenticated: func(connection *connectionT) { - username := connection.user.username - rwmutex.Lock() - defer rwmutex.Unlock() - if state.users[username] == nil { - state.users[username] = []uuid.UUID{} - } - state.users[username] = append( - state.users[username], - connection.uuid, - ) - }, - */ - subscribe: func( - username string, - channelNames []string, - ) { - rwmutex.Lock() - defer rwmutex.Unlock() - for _, channelName := range channelNames { - if state.members[channelName] == nil { - state.members[channelName] = - map[string][]uuid.UUID{} - } - state.members[channelName][username] = - state.users[username] - } - }, - /* - members: func(channelName string) []string { - rwmutex.RLock() - defer rwmutex.RUnlock() - usernames := make( - []string, - len(state.members[channelName]), - ) - i := 0 - for username, _ := range state.members[channelName] { - usernames[i] = username - i++ - } - return usernames - }, - connections: func(username string) []uuid.UUID { - rwmutex.RLock() - defer rwmutex.RUnlock() - connections := make( - []uuid.UUID, - len(state.users[username]), - ) - copy(connections, state.users[username]) - return connections - }, - connection: func(connectionID uuid.UUID) *connectionT { - rwmutex.RLock() - defer rwmutex.RUnlock() - return state.connections[connectionID] - }, - */ - } -} - -func newState() *stateT { - return &stateT{ +func newState() stateT { + return stateT{ members: pds.NewMap[string, []string](nil), users: pds.NewMap[string, []uuid.UUID](nil), connections: pds.NewMap[uuid.UUID, connectionT](nil), } } -func buildMetrics(prefix string) metricsT { +func buildMetrics(tag string) metricsT { return metricsT{ activeConnections: g.MakeGauge( "active-connection", - "prefix", prefix, + "tag", tag, ), nicksInChannel: g.MakeGauge( "nicks-in-channel", - "prefix", prefix, + "tag", tag, ), sendToClientError: g.MakeCounter( "send-to-client-error", - "prefix", prefix, + "tag", tag, ), receivedMessage: g.MakeCounter( "received-message", - "prefix", prefix, + "tag", tag, ), sentReply: g.MakeCounter( "sent-reply", - "prefix", prefix, + "tag", tag, ), } } -func NewWithPrefix( - databasePath string, - prefix string, - daemonSocketPath string, - commanderSocketPath string, -) (IPapod, error) { - queue, err := fiinha.New(databasePath) - if err != nil { - return papodT{}, err - } +func initDB(path string) (acude.AcudeI, error) { + return nil, nil +} - auth, err := cracha.New(databasePath) +func newPapod(baseDir string, tag string) (papodT, error) { + databasePath := baseDir + "/papod.dedo" + daemonSocketPath := baseDir + "/papod.daemon.socket" + commanderSocketPath := baseDir + "/papod.commander.socket" + + cracha, err := cracha.New(databasePath) if err != nil { - queue.Close() return papodT{}, err } listeners, err := initListeners(daemonSocketPath, commanderSocketPath) if err != nil { - queue.Close() - auth.Close() + cracha.Close() return papodT{}, err } - queries, err := initDB(databasePath, prefix) + acude, err := initDB(databasePath) if err != nil { - queue.Close() - auth.Close() + cracha.Close() listeners.close() return papodT{}, err } - consumers := buildConsumers(prefix) - stateMutable := newStateMutable() - state := newState() - // receivers := makeReceivers() - metrics := buildMetrics(prefix) - // logger := g.NewLogger("prefix", prefix, "program", "papod") - return papodT{ - queue: queue, - auth: auth, - queries: queries, + acude: acude, + cracha: cracha, listeners: listeners, - consumers: consumers, - stateMutable: stateMutable, - state: stm.Atom(state), - // receivers: receivers, - metrics: metrics, - // logger: logger, + state: stm.Atom(newState()), + metrics: buildMetrics(tag), }, nil } -func New(basePath string) (IPapod, error) { - return NewWithPrefix( - basePath + "/papod.db", - defaultPrefix, - basePath + "/papod.daemon.sock", - basePath + "/papod.commander.sock", - ) +func NewWith(baseDir string, tag string) (PapoI, error) { + return newPapod(baseDir, tag) +} + +func New() (PapoI, error) { + return NewWith(".", "papod-default") } func splitOnCRLF(data []byte, _atEOF bool) (int, []byte, error) { @@ -3796,24 +303,8 @@ func parseMessage(rawMessage string) (messageT, error) { return msg, nil } -func asNewEvent(msg messageT) newEventT { - return newEventT{} -} - -func joinEvent(member memberT) eventT { - return eventT{} -} - -func asMessage(event eventT) messageT { - return messageT{} -} - -func asReply(event eventT) replyT { - return replyT{} -} - -func removeConnection(state *stateT, connection *connectionT) *stateT { - return state +func removeConnection(state stateT, connection *connectionT) stateT { + return state // FIXME } /// Is this death by a thousand goroutines? Is the runtime able to handle the @@ -3824,7 +315,7 @@ func removeConnection(state *stateT, connection *connectionT) *stateT { func broadcastMessage( message messageT, channelName string, - state *stateT, + state stateT, ) { usernames, _ := state.members.Get(channelName) for _, username := range usernames { @@ -4147,8 +638,8 @@ func handleJOIN( msg messageT, ) ([]replyT, bool, error) { // FIXME: add to database - channels := strings.FieldsFunc(msg.params[0], splitCommas) - papod.stateMutable.subscribe(connection.user.username, channels) + // channels := strings.FieldsFunc(msg.params[0], splitCommas) + // papod.stateMutable.subscribe(connection.user.username, channels) return []replyT{ _RPL_NOTOPIC (connection, msg), @@ -4189,7 +680,6 @@ func handleAWAY( msg messageT, ) ([]replyT, bool, error) { replyFn := _RPL_NOWAWAY - if len(msg.params) == 0 { replyFn = _RPL_UNAWAY } @@ -4304,6 +794,7 @@ func handleUnknown( msg messageT, ) ([]replyT, bool, error) { // FIXME: user doesn't exist when unauthenticated + /* err := papod.queries.logMessage(userT{ }, msg) if err != nil { g.Warning( @@ -4314,6 +805,7 @@ func handleUnknown( "err", err, ) } + */ return []replyT{ _ERR_UNKNOWNCOMMAND(connection, msg), @@ -4422,7 +914,7 @@ func processMessage( ) } - stm.Swap(papod.state, func(state *stateT) *stateT { + stm.Swap(papod.state, func(state stateT) stateT { return removeConnection(state, connection) }) err := connection.conn.Close() @@ -4439,10 +931,13 @@ func processMessage( } if shouldClose { + // FIXME // papod.stateMutable.disconnect(connection) } } +func processTasks() // FIXME + func handleConnection(papod papodT, conn net.Conn) { connection := connectionT{ uuid: uuid.New(), @@ -4509,27 +1004,17 @@ func mkbgrun() (func(func()), func()) { } func (papod papodT) Start() error { - err := registerConsumers( - papod.queue.Subscribe, - papod.queue.Unsubscribe, - papod, - papod.consumers, - ) - if err != nil { - return err - } - - // FIXME: papod.logger.Info g.Info("Starting service", "lifecycle-event", "event", "starting-server", slog.Group( "versions", - "gobang", g.Version, "cracha", cracha.Version, - "fiinha", fiinha.Version, - "golite", golite.Version, - "uuid", uuid.Version, + "uuid", uuid.Version, + "pds", pds.Version, + "stm", stm.Version, "papod", Version, + "gobang", g.Version, + "gotext", gt.Version, ), ) @@ -4543,7 +1028,7 @@ func (papod papodT) Start() error { func (papod papodT) Close() error { // FIXME: does this wait for current handlers to wait? Well, it should. - unregisterConsumers(papod.queue.Unsubscribe, papod.consumers) + /* return g.WrapErrors( papod.listeners.close(), // papod.connCloser.closeAll(), @@ -4551,36 +1036,76 @@ func (papod papodT) Close() error { papod.queue.Close(), papod.queries.close(), ) + */ + return nil } -func basePathFrom(args []string) (string, error) { - if len(args) < 2 { - return os.Getwd() - } else { - return args[1], nil - } +func usage(argv0 string, w io.Writer) { + fmt.Fprintf( + w, + gt.Gettext("Usage: %s [-t TAG] [BASEDIR]"), + argv0, + ) } +func getopt(allArgs []string, w io.Writer) (argsT, int) { + argv0 := allArgs[0] + argv := allArgs[1:] + fs := flag.NewFlagSet("", flag.ContinueOnError) + fs.Usage = func() {} + fs.SetOutput(w) + tag := fs.String( + "t", + "", + "The specific instance tag for inclusion in the log", + ) + if fs.Parse(argv) != nil { + usage(argv0, w) + return argsT{}, 2 + } -func Main() { - g.Init("program", "papod") - - basePath, err := basePathFrom(os.Args) - if err != nil { - fmt.Printf("%v\n", err) - os.Exit(1) + subArgs := fs.Args() + baseDir := "." + if len(subArgs) != 0 { + baseDir = subArgs[0] } - ipapod, err := New(basePath) + return argsT{ + allArgs: allArgs, + subArgs: subArgs, + baseDir: baseDir, + tag: *tag, + }, 0 +} + +func run(env envT) int { + papod, err := newPapod(env.args.baseDir, env.args.tag) if err != nil { - fmt.Printf("Failed to create papod: %v\n", err) - os.Exit(1) + fmt.Fprintln(env.err, err) + return 1 } - err = ipapod.Start() + err = papod.Start() if err != nil { - fmt.Printf(gt.Gettext("Failed to start: %v\n"), err) - os.Exit(1) + fmt.Fprintln(env.err, err) + return 1 } + + return 0 +} + + + +func Main() { + g.Init() + gt.Init(Name, LOCALEDIR) + args, rc := getopt(os.Args, os.Stderr) + g.ExitIf(rc) + os.Exit(run(envT{ + args: args, + in: os.Stdin, + out: os.Stdout, + err: os.Stderr, + })) } |