Changes
11 changed files (+314/-118)
-
-
@@ -27,8 +27,8 @@ You will be in trouble if you use part or whole of this application in a real pr## Architecture The simplest description is "Snapshot-less Event Sourcing HTTP server using SQLite3 as an event store". Everytime HTTP handler needs the current state of something (user list, one-time password for initial admin creation), the application loads every events then builds state by iterating events one-by-one. The simplest description is "Event Sourcing HTTP server using SQLite3 as an event store". Everytime HTTP handler needs the current state of something (user list, one-time password for initial admin creation), the application loads latest projection snapshot and events newer than the snapshot, then builds state from those. Events are stored in SQLite3 table named `user_events` with dead simple schema:
-
@@ -40,13 +40,26 @@ Events are stored in SQLite3 table named `user_events` with dead simple schema:`seq` is auto incrementing sequential number. `payload` is binary data in Protobuf wire format. `event_name` is schema name of the `payload`, telling which Protobuf message to use for decoding. `event_name` is schema name of the `payload`, telling which Protobuf message schema to use for decoding. As this is demo application, `event_name` does not contain package name. Table schema is defined in `init.sql` file and insertion/retrieval logics are inside `events/` directory. Protobuf message schemas are under `proto/` directory. To reduce the number of events loaded, the latest projection is saved as a snapshot in `users_snapshot` table, which has the following schema: Once every events are loaded, the application constructs current state from those events. These logics are in `projections/` directory and they have unit tests. | Column | Data type | | ----------- | --------- | | `event_seq` | `INTEGER` | | `payload` | `BLOB` | `event_seq` is a `seq` column of the event this snapshot was created at. `payload` is binary data in Protobuf wire format, same as `user_events`. Schema for snapshots is under `proto/projection` directory. Projections, unlike events, has dedicated table for each projection types. Therefore `*_snapshot` table does not have a column to store Protobuf schema name. Table schema is defined in `init.sql` file. Protobuf message schemas for events payload are under `proto/event/` directory and ones for projections/snapshots are under `proto/projection/` directory. Events listing and insertion functions are in `events/` directory. Functions to build projection from events and a snapshot are inside `projections/` directory and they have unit tests. Creation of demo users and one-time password is defined in `setups/` directory. This directory is good candidate of unit testing but I'm lazy so there's none.
-
-
-
@@ -37,7 +37,7 @@ func List(db *sql.DB) ([]proto.Message, error) {return []proto.Message{}, nil } rows, err := tx.Query("SELECT event_name, payload FROM user_events ORDER BY seq ASC") rows, err := tx.Query("SELECT seq, event_name, payload FROM user_events ORDER BY seq ASC") if err != nil { return nil, fmt.Errorf("Failed to SELECT user_events: %s", err) }
-
@@ -48,41 +48,55 @@ func List(db *sql.DB) ([]proto.Message, error) {return nil, fmt.Errorf("Number of events is less than rowCount") } var eventName string var payload []byte if err := rows.Scan(&eventName, &payload); err != nil { return nil, fmt.Errorf("Failed to scan user event: %s", err) event, _, err := ScanEvent(rows) if err != nil { return nil, err } switch eventName { case "InitialAdminCreationPasswordCreated": var event event.InitialAdminCreationPasswordCreated if err := proto.Unmarshal(payload, &event); err != nil { return nil, fmt.Errorf("Illegal InitialAdminCreationPasswordCreated event: %s", err) } events[i] = &event case "UserCreated": var event event.UserCreated if err := proto.Unmarshal(payload, &event); err != nil { return nil, fmt.Errorf("Illegal UserCreated event: %s", err) } events[i] = &event case "PasswordLoginConfigured": var event event.PasswordLoginConfigured if err := proto.Unmarshal(payload, &event); err != nil { return nil, fmt.Errorf("Illegal PasswordLoginConfigured event: %s", err) } events[i] = &event case "RoleAssigned": var event event.RoleAssigned if err := proto.Unmarshal(payload, &event); err != nil { return nil, fmt.Errorf("Illegal RoleAssigned event: %s", err) } events[i] = &event default: return nil, fmt.Errorf("Unknown event in user_events: name=%s", eventName) } events[i] = event } return events, nil } type Scanner interface { Scan(dst ...any) error } func ScanEvent(scanner Scanner) (proto.Message, int, error) { var seq int var eventName string var payload []byte if err := scanner.Scan(&seq, &eventName, &payload); err != nil { return nil, 0, fmt.Errorf("Failed to scan user event: %s", err) } switch eventName { case "InitialAdminCreationPasswordCreated": var event event.InitialAdminCreationPasswordCreated if err := proto.Unmarshal(payload, &event); err != nil { return nil, 0, fmt.Errorf("Illegal InitialAdminCreationPasswordCreated event: %s", err) } return &event, seq, nil case "UserCreated": var event event.UserCreated if err := proto.Unmarshal(payload, &event); err != nil { return nil, 0, fmt.Errorf("Illegal UserCreated event: %s", err) } return &event, seq, nil case "PasswordLoginConfigured": var event event.PasswordLoginConfigured if err := proto.Unmarshal(payload, &event); err != nil { return nil, 0, fmt.Errorf("Illegal PasswordLoginConfigured event: %s", err) } return &event, seq, nil case "RoleAssigned": var event event.RoleAssigned if err := proto.Unmarshal(payload, &event); err != nil { return nil, 0, fmt.Errorf("Illegal RoleAssigned event: %s", err) } return &event, seq, nil default: return nil, 0, fmt.Errorf("Unknown event in user_events: name=%s", eventName) } }
-
-
-
@@ -12,3 +12,10 @@ CREATE TABLE user_events (event_name TEXT NOT NULL ON CONFLICT ROLLBACK, payload BLOB ); CREATE TABLE users_snapshots ( -- Which event is this snapshot taken at? event_seq INTEGER PRIMARY KEY ON CONFLICT ROLLBACK, -- Protobuf wire format payload BLOB );
-
-
-
@@ -10,66 +10,126 @@package users import ( "context" "database/sql" "fmt" "google.golang.org/protobuf/proto" "pocka.jp/x/event_sourcing_user_management_poc/events" "pocka.jp/x/event_sourcing_user_management_poc/gen/event" "pocka.jp/x/event_sourcing_user_management_poc/gen/model" "pocka.jp/x/event_sourcing_user_management_poc/gen/projection" ) type passwordLogin struct { Hash []byte Salt []byte } func GetProjection(db *sql.DB) (*projection.UsersProjection, int, error) { ctx := context.Background() var p projection.UsersProjection tx, err := db.BeginTx(ctx, nil) if err != nil { return nil, 0, fmt.Errorf("Failed to begin transaction for UsersProjection: %s", err) } defer tx.Rollback() var eventSeq int var payload []byte err = tx.QueryRow("SELECT event_seq, payload FROM users_snapshots ORDER BY event_seq DESC LIMIT 1").Scan(&eventSeq, &payload) if err == sql.ErrNoRows { p = projection.UsersProjection{ Users: []*projection.User{}, } eventSeq = -1 } else if err != nil { return nil, 0, fmt.Errorf("Failed to get latest snapshot: %s", err) } else { if err := proto.Unmarshal(payload, &p); err != nil { return nil, 0, fmt.Errorf("Failed to decode latest snapshot: %s", err) } } stmt, err := tx.Prepare("SELECT seq, event_name, payload FROM user_events WHERE seq > ? ORDER BY seq ASC") if err != nil { return nil, 0, fmt.Errorf("Failed to prepare event fetching query: %s", err) } maxSeq := -1 rows, err := stmt.Query(eventSeq) for rows.Next() { ev, seq, err := events.ScanEvent(rows) if err != nil { return nil, 0, err } type user struct { ID string DisplayName string Email string maxSeq = max(maxSeq, seq) PasswordLogin *passwordLogin Role *model.Role apply(ev, &p) } return &p, maxSeq, nil } func ListFromUserEvents(events []proto.Message) []user { users := make(map[string]*user) func apply(ev proto.Message, p *projection.UsersProjection) { switch v := ev.(type) { case *event.UserCreated: p.Users = append(p.Users, &projection.User{ Id: v.Id, DisplayName: v.DisplayName, Email: v.Email, }) return case *event.PasswordLoginConfigured: if v.UserId == nil { return } for _, e := range events { switch v := e.(type) { case *event.UserCreated: users[*v.Id] = &user{ ID: *v.Id, DisplayName: *v.DisplayName, Email: *v.Email, } case *event.PasswordLoginConfigured: if v.UserId == nil { break for _, user := range p.Users { if *user.Id != *v.UserId { continue } found := users[*v.UserId] if found != nil { found.PasswordLogin = &passwordLogin{ Hash: v.PasswordHash, Salt: v.Salt, } } case *event.RoleAssigned: if v.UserId == nil { break user.PasswordLogin = &projection.User_PasswordLogin{ Hash: v.PasswordHash, Salt: v.Salt, } return } return case *event.RoleAssigned: if v.UserId == nil { return } found := users[*v.UserId] if found != nil { found.Role = v.Role for _, user := range p.Users { if *user.Id != *v.UserId { continue } user.Role = v.Role return } return } } func SaveSnapshot(db *sql.DB) error { p, seq, err := GetProjection(db) if err != nil { return err } ret := make([]user, 0, len(users)) stmt, err := db.Prepare("INSERT OR ABORT INTO users_snapshots (event_seq, payload) VALUES (?, ?)") if err != nil { return err } for _, u := range users { ret = append(ret, *u) payload, err := proto.Marshal(p) if err != nil { return err } return ret _, err = stmt.Exec(seq, payload) return err }
-
-
-
@@ -17,10 +17,21 @@ import ("pocka.jp/x/event_sourcing_user_management_poc/gen/event" "pocka.jp/x/event_sourcing_user_management_poc/gen/model" "pocka.jp/x/event_sourcing_user_management_poc/gen/projection" ) func build(events []proto.Message) *projection.UsersProjection { var p projection.UsersProjection for _, e := range events { apply(e, &p) } return &p } func TestIdentityOnly(t *testing.T) { users := ListFromUserEvents([]proto.Message{ p := build([]proto.Message{ &event.UserCreated{ Id: proto.String("foo"), DisplayName: proto.String("Foo"),
-
@@ -32,33 +43,33 @@ func TestIdentityOnly(t *testing.T) {}, }) if len(users) != 1 { t.Errorf("Expected 1 user, got %d", len(users)) if len(p.Users) != 1 { t.Errorf("Expected 1 user, got %d", len(p.Users)) } if users[0].ID != "foo" { t.Errorf("Expected ID \"foo\", got \"%s\"", users[0].ID) if *p.Users[0].Id != "foo" { t.Errorf("Expected ID \"foo\", got \"%s\"", *p.Users[0].Id) } if users[0].DisplayName != "Foo" { t.Errorf("Expected DisplayName \"Foo\", got \"%s\"", users[0].DisplayName) if *p.Users[0].DisplayName != "Foo" { t.Errorf("Expected DisplayName \"Foo\", got \"%s\"", *p.Users[0].DisplayName) } if users[0].Email != "foo@example.com" { t.Errorf("Expected Email \"foo@example.com\", got \"%s\"", users[0].Email) if *p.Users[0].Email != "foo@example.com" { t.Errorf("Expected Email \"foo@example.com\", got \"%s\"", *p.Users[0].Email) } if users[0].Role != nil { t.Errorf("Expected Role to be nil, got %v", users[0].Role) if p.Users[0].Role != nil { t.Errorf("Expected Role to be nil, got %v", p.Users[0].Role) } if users[0].PasswordLogin != nil { t.Errorf("Expected Role to be nil, got %v", users[0].Role) if p.Users[0].PasswordLogin != nil { t.Errorf("Expected Role to be nil, got %v", p.Users[0].Role) } } func TestWithRole(t *testing.T) { users := ListFromUserEvents([]proto.Message{ p := build([]proto.Message{ &event.UserCreated{ Id: proto.String("foo"), DisplayName: proto.String("Foo"),
-
@@ -70,13 +81,13 @@ func TestWithRole(t *testing.T) {}, }) if *users[0].Role != model.Role_ROLE_ADMIN { t.Errorf("Expected Role_ROLE_ADMIN, got %v", users[0].Role.String()) if *p.Users[0].Role != model.Role_ROLE_ADMIN { t.Errorf("Expected Role_ROLE_ADMIN, got %v", p.Users[0].Role.String()) } } func TestWithPWLogin(t *testing.T) { users := ListFromUserEvents([]proto.Message{ p := build([]proto.Message{ &event.UserCreated{ Id: proto.String("foo"), DisplayName: proto.String("Foo"),
-
@@ -89,11 +100,11 @@ func TestWithPWLogin(t *testing.T) {}, }) if !bytes.Equal(users[0].PasswordLogin.Hash, []byte{0, 1, 2}) { t.Errorf("Expected [0,1,2], got %v", users[0].PasswordLogin.Hash) if !bytes.Equal(p.Users[0].PasswordLogin.Hash, []byte{0, 1, 2}) { t.Errorf("Expected [0,1,2], got %v", p.Users[0].PasswordLogin.Hash) } if !bytes.Equal(users[0].PasswordLogin.Salt, []byte{3, 4, 5}) { t.Errorf("Expected [3,4,5], got %v", users[0].PasswordLogin.Salt) if !bytes.Equal(p.Users[0].PasswordLogin.Salt, []byte{3, 4, 5}) { t.Errorf("Expected [3,4,5], got %v", p.Users[0].PasswordLogin.Salt) } }
-
-
-
@@ -0,0 +1,29 @@// Copyright 2025 Shota FUJI // // This source code is licensed under Zero-Clause BSD License. // You can find a copy of the Zero-Clause BSD License at LICENSES/0BSD.txt // You may also obtain a copy of the Zero-Clause BSD License at // <https://opensource.org/license/0bsd> // // SPDX-License-Identifier: 0BSD edition = "2023"; package projection; import "proto/model/role.proto"; option go_package = "pocka.jp/x/event_sourcing_user_management_poc/gen/projection"; message User { string id = 1; string display_name = 2; string email = 3; PasswordLogin password_login = 4; model.Role role = 5; message PasswordLogin { bytes hash = 1; bytes salt = 2; } }
-
-
-
@@ -0,0 +1,20 @@// Copyright 2025 Shota FUJI // // This source code is licensed under Zero-Clause BSD License. // You can find a copy of the Zero-Clause BSD License at LICENSES/0BSD.txt // You may also obtain a copy of the Zero-Clause BSD License at // <https://opensource.org/license/0bsd> // // SPDX-License-Identifier: 0BSD edition = "2023"; package projection; import "proto/projection/user.proto"; option go_package = "pocka.jp/x/event_sourcing_user_management_poc/gen/projection"; message UsersProjection { repeated User users = 1; }
-
-
-
@@ -73,13 +73,19 @@ func Handler(db *sql.DB, logger *log.Logger) (http.Handler, error) {return } users := users.ListFromUserEvents(events) p, _, err := users.GetProjection(db) if err != nil { w.Header().Add("Content-Type", "text/html;charset=utf-8") w.WriteHeader(http.StatusInternalServerError) fmt.Fprint(w, loginHTML) return } for _, user := range users { for _, user := range p.Users { // No real auth. No security. if user.ID == id.Value { if *user.Id == id.Value { loggedInAdminHtml.Execute(w, loggedInAdminPipeline{ DisplayName: user.DisplayName, DisplayName: *user.DisplayName, Role: user.Role.String(), }) return
-
@@ -157,6 +163,17 @@ func Handler(db *sql.DB, logger *log.Logger) (http.Handler, error) {return } go func() { logger.Debug("Creating snapshot (trigger=initial admin creation)") err := users.SaveSnapshot(db) if err != nil { logger.Errorf("Failed to update users snapshot: %s", err) } logger.Debug("Created snapshot (trigger=initial admin creation)") }() // This project is PoC for event sourcing. UI and security is completely out-of-scope. http.SetCookie(w, &http.Cookie{ Name: "id",
-
@@ -169,15 +186,14 @@ func Handler(db *sql.DB, logger *log.Logger) (http.Handler, error) {mux.HandleFunc("/login", func(w http.ResponseWriter, r *http.Request) { r.ParseForm() events, err := events.List(db) p, _, err := users.GetProjection(db) if err != nil { logger.Error(err) http.Error(w, "Server error: event loading failure", http.StatusInternalServerError) w.Header().Add("Content-Type", "text/html;charset=utf-8") w.WriteHeader(http.StatusInternalServerError) fmt.Fprint(w, loginHTML) return } users := users.ListFromUserEvents(events) email := r.PostForm.Get("email") password := r.PostForm.Get("password")
-
@@ -187,15 +203,15 @@ func Handler(db *sql.DB, logger *log.Logger) (http.Handler, error) {return } for _, user := range users { for _, user := range p.Users { // No real auth. No security. if user.Email == email && user.PasswordLogin != nil { if *user.Email == email && user.PasswordLogin != nil { hash := auth.HashPassword(password, user.PasswordLogin.Salt) if bytes.Equal(user.PasswordLogin.Hash, hash) { // This project is PoC for event sourcing. UI and security is completely out-of-scope. http.SetCookie(w, &http.Cookie{ Name: "id", Value: user.ID, Value: *user.Id, }) http.Redirect(w, r, "/", http.StatusFound)
-
-
-
@@ -118,7 +118,7 @@ func main() {if *shouldCreateAlice { logger.Debug("Creating admin user Alice...") id, err := setups.CreateAlice(db) id, err := setups.CreateAlice(db, logger) if err != nil { logger.Fatal(err) }
-
@@ -129,7 +129,7 @@ func main() {if *shouldCreateBob { logger.Debug("Creating viewer user Bob...") id, err := setups.CreateBob(db) id, err := setups.CreateBob(db, logger) if err != nil { logger.Fatal(err) }
-
-
-
@@ -13,6 +13,7 @@ import ("database/sql" "fmt" "github.com/charmbracelet/log" "github.com/google/uuid" "google.golang.org/protobuf/proto"
-
@@ -20,12 +21,13 @@ import ("pocka.jp/x/event_sourcing_user_management_poc/events" "pocka.jp/x/event_sourcing_user_management_poc/gen/event" "pocka.jp/x/event_sourcing_user_management_poc/gen/model" "pocka.jp/x/event_sourcing_user_management_poc/projections/users" ) // CreateAlice creates a new admin user named "Alice" with demo password of // "Alice's password". // CreateAlice returns an ID of the created user on success. func CreateAlice(db *sql.DB) (string, error) { func CreateAlice(db *sql.DB, logger *log.Logger) (string, error) { id := uuid.New().String() passwordHash, salt := auth.HashPasswordWithRandomSalt("Alice's password")
-
@@ -49,5 +51,16 @@ func CreateAlice(db *sql.DB) (string, error) {return "", fmt.Errorf("Unable to create Alice: %s", err) } go func() { logger.Debug("Creating snapshot (trigger=create alice)") err := users.SaveSnapshot(db) if err != nil { logger.Warnf("Failed to create user snapshot: %s", err) } logger.Debug("Created snapshot (trigger=create alice)") }() return id, nil }
-
-
-
@@ -13,6 +13,7 @@ import ("database/sql" "fmt" "github.com/charmbracelet/log" "github.com/google/uuid" "google.golang.org/protobuf/proto"
-
@@ -20,12 +21,13 @@ import ("pocka.jp/x/event_sourcing_user_management_poc/events" "pocka.jp/x/event_sourcing_user_management_poc/gen/event" "pocka.jp/x/event_sourcing_user_management_poc/gen/model" "pocka.jp/x/event_sourcing_user_management_poc/projections/users" ) // CreateBob creates a new viewer user named "Bob" with demo password of // "Bob's password". // CreateBob returns an ID of the created user on success. func CreateBob(db *sql.DB) (string, error) { func CreateBob(db *sql.DB, logger *log.Logger) (string, error) { id := uuid.New().String() passwordHash, salt := auth.HashPasswordWithRandomSalt("Bob's password")
-
@@ -49,5 +51,16 @@ func CreateBob(db *sql.DB) (string, error) {return "", fmt.Errorf("Unable to create Bob: %s", err) } go func() { logger.Debug("Creating snapshot (trigger=create bob)") err := users.SaveSnapshot(db) if err != nil { logger.Warnf("Failed to create user snapshot: %s", err) } logger.Debug("Created snapshot (trigger=create bob)") }() return id, nil }
-