From 6082e59d47c17a9d54a6b0556b3f31559d7c620a Mon Sep 17 00:00:00 2001 From: Silvan <27845747+adlerhurst@users.noreply.github.com> Date: Mon, 15 Jun 2026 11:24:37 +0200 Subject: [PATCH] fix(eventstore): allow overwriting resource owner of events (#12261) # Which Problems Are Solved - The eventstore did not support intentionally overwriting the resource owner when creating events for aggregates that may be reused across owners. - Resource owner handling was implicit and could not be controlled per command/event type. - We needed a safe way to distinguish between: - keeping the existing aggregate owner, and - explicitly setting a new owner for specific create-like events. # How the Problems Are Solved - Introduced a new eventstore command type with an explicit enforce_owner flag. - Updated eventstore.commands_to_events and eventstore.push so owner assignment is now explicit: - if enforce_owner is true, the command owner is written - if enforce_owner is false, the existing aggregate owner is retained when present - Added EnforceResourceOwnerCommand and wiring so command types can opt in to enforced owner behavior. - Wired the new behavior through the v3 eventstore push path, including compatibility fallback for older command type mapping. - Added migration/setup changes to register and use the new command type and SQL functions. - Added and updated tests for owner overwrite and aggregate ID reuse scenarios. # Additional Changes - Added small migration/setup robustness improvements related to eventstore setup ordering and helper reuse. - Added focused test coverage for enforced owner behavior and sequencing. - Events that currently allow owner changes (implement EnforceResourceOwner) are: - AddedEvent (action) - GroupAddedEvent - StartedEvent (idp intent) - ProjectAddedEvent - HumanAddedEvent - HumanRegisteredEvent - MachineAddedEvent - CreatedEvent (schema user) # Additional Context - Follow-up for eventstore owner-handling correctness in create flows and aggregate ID reuse cases. - No additional issue link was attached for this change. --------- Co-authored-by: abhishek kumar gupta --- cmd/initialise/sql/08_events_table.sql | 183 +++++++++++------- cmd/setup/40/01_type.sql | 3 +- cmd/setup/40/02_func.sql | 178 ++++++++++------- cmd/setup/64.go | 6 +- cmd/setup/64.sql | 150 ++++++++++---- cmd/setup/70.go | 34 ++++ cmd/setup/70.sql | 136 +++++++++++++ cmd/setup/config.go | 1 + cmd/setup/setup.go | 2 + internal/command/group_model.go | 4 + internal/command/metadata_model.go | 9 + internal/command/org_action_model.go | 22 ++- internal/command/org_domain_model.go | 35 +++- internal/command/org_metadata_model.go | 8 + .../command/project_application_api_model.go | 15 ++ .../command/project_application_key_model.go | 13 ++ internal/command/project_application_model.go | 7 +- .../command/project_application_oidc_model.go | 50 ++++- .../command/project_application_saml_model.go | 17 ++ .../project_application_secret_model.go | 11 ++ .../command/project_grant_member_model.go | 5 + internal/command/project_grant_model.go | 7 +- internal/command/project_model.go | 5 + internal/command/project_role_model.go | 5 +- internal/command/session_test.go | 6 +- internal/command/user_grant_model.go | 2 + .../command/user_human_access_token_model.go | 26 +++ internal/command/user_human_address_model.go | 13 ++ internal/command/user_human_email_model.go | 27 +++ internal/command/user_human_init_model.go | 25 +++ internal/command/user_human_model.go | 43 ++++ internal/command/user_human_otp_model.go | 28 ++- internal/command/user_human_password_model.go | 31 +++ internal/command/user_human_phone_model.go | 42 +++- internal/command/user_human_profile_model.go | 18 +- .../command/user_human_recovery_code_model.go | 14 +- .../user_human_recovery_code_model_test.go | 2 +- .../command/user_human_refresh_token_model.go | 21 ++ internal/command/user_human_webauthn_model.go | 87 +++++++-- internal/command/user_idp_link_model.go | 12 +- internal/command/user_machine_key_model.go | 15 +- internal/command/user_machine_model.go | 17 +- internal/command/user_metadata_model.go | 20 +- internal/command/user_model.go | 6 + internal/command/user_v2_invite_model.go | 30 ++- internal/command/user_v2_model.go | 46 +++++ internal/command/user_v2_model_test.go | 22 +-- internal/eventstore/event.go | 11 ++ internal/eventstore/eventstore_pusher_test.go | 91 +++++++++ internal/eventstore/local_postgres_test.go | 38 +++- internal/eventstore/v3/event.go | 2 + internal/eventstore/v3/event_test.go | 74 +++++++ internal/eventstore/v3/eventstore.go | 110 +++++++++-- internal/eventstore/v3/mock_test.go | 7 + internal/eventstore/v3/push.go | 7 +- internal/eventstore/v3/sequence.go | 9 +- internal/eventstore/v3/sequence_test.go | 39 ++++ internal/eventstore/write_model.go | 6 + internal/query/user_otp.go | 1 + internal/query/user_password.go | 23 +++ internal/repository/action/action.go | 2 + internal/repository/group/group.go | 2 + internal/repository/project/project.go | 2 + internal/repository/user/human.go | 4 + internal/repository/user/machine.go | 2 + internal/repository/user/schemauser/user.go | 2 + 66 files changed, 1603 insertions(+), 288 deletions(-) create mode 100644 cmd/setup/70.go create mode 100644 cmd/setup/70.sql diff --git a/cmd/initialise/sql/08_events_table.sql b/cmd/initialise/sql/08_events_table.sql index c4b7351e6..c22394838 100644 --- a/cmd/initialise/sql/08_events_table.sql +++ b/cmd/initialise/sql/08_events_table.sql @@ -23,7 +23,7 @@ CREATE INDEX IF NOT EXISTS es_projection ON eventstore.events2 (instance_id, agg -- represents an event to be created. DO $$ BEGIN - CREATE TYPE eventstore.command AS ( + CREATE TYPE eventstore.command2 AS ( instance_id TEXT , aggregate_type TEXT , aggregate_id TEXT @@ -32,90 +32,131 @@ DO $$ BEGIN , payload JSONB , creator TEXT , owner TEXT + , enforce_owner BOOLEAN ); EXCEPTION WHEN duplicate_object THEN null; END $$; -CREATE OR REPLACE FUNCTION eventstore.commands_to_events(commands eventstore.command[]) RETURNS SETOF eventstore.events2 VOLATILE AS $$ -SELECT - c.instance_id - , c.aggregate_type - , c.aggregate_id - , c.command_type AS event_type - , cs.sequence + ROW_NUMBER() OVER (PARTITION BY c.instance_id, c.aggregate_type, c.aggregate_id ORDER BY c.in_tx_order) AS sequence - , c.revision - , NOW() AS created_at - , c.payload - , c.creator - , cs.owner - , EXTRACT(EPOCH FROM NOW()) AS position - , c.in_tx_order -FROM ( - SELECT +DO $$ BEGIN + CREATE TYPE eventstore.latest_command AS ( + instance_id TEXT + , aggregate_type TEXT + , aggregate_id TEXT + , owner TEXT + , sequence BIGINT + ); +EXCEPTION + WHEN duplicate_object THEN null; +END $$; + +CREATE OR REPLACE FUNCTION eventstore.commands_to_events(commands eventstore.command2[]) + RETURNS SETOF eventstore.events2 + LANGUAGE 'plpgsql' + STABLE PARALLEL SAFE + ROWS 10 +AS $$ +DECLARE + _latest_events eventstore.latest_command[]; + command eventstore.command2; + last_owner TEXT; + i INTEGER; +BEGIN + + SELECT array_agg(res) INTO _latest_events + FROM ( + SELECT + ROW( + c.instance_id, + c.aggregate_type, + c.aggregate_id, + COALESCE(e.owner, c.owner), + COALESCE(e.sequence, 0) + )::eventstore.latest_command AS res + FROM ( + SELECT DISTINCT ON (c.instance_id, c.aggregate_type, c.aggregate_id) + c.instance_id, + c.aggregate_type, + c.aggregate_id, + c.owner, + c.ordinality + FROM ( + SELECT * + FROM UNNEST(commands) WITH ORDINALITY AS c + ) AS c + ORDER BY + c.instance_id, + c.aggregate_type, + c.aggregate_id, + c.ordinality + ) AS c + LEFT JOIN LATERAL ( + SELECT + e.owner, + e.sequence + FROM eventstore.events2 AS e + WHERE e.instance_id = c.instance_id + AND e.aggregate_type = c.aggregate_type + AND e.aggregate_id = c.aggregate_id + ORDER BY e.sequence DESC + LIMIT 1 + ) AS e ON TRUE + ) AS latest; + + FOR command IN SELECT * FROM UNNEST(commands) LOOP + IF NOT command.enforce_owner THEN + SELECT owner INTO command.owner + FROM UNNEST(_latest_events) AS e + WHERE e.instance_id = command.instance_id + AND e.aggregate_type = command.aggregate_type + AND e.aggregate_id = command.aggregate_id; + ELSE + FOR i IN 1..array_length(_latest_events, 1) LOOP + IF _latest_events[i].instance_id = command.instance_id + AND _latest_events[i].aggregate_type = command.aggregate_type + AND _latest_events[i].aggregate_id = command.aggregate_id + THEN + _latest_events[i].owner := command.owner; + EXIT; + END IF; + END LOOP; + END IF; + + END LOOP; + + + RETURN QUERY SELECT c.instance_id , c.aggregate_type , c.aggregate_id - , c.command_type + , c.command_type AS event_type + , COALESCE(e.sequence, 0) + ROW_NUMBER() OVER (PARTITION BY c.instance_id, c.aggregate_type, c.aggregate_id ORDER BY c.in_tx_order) AS sequence , c.revision + , NOW() AS created_at , c.payload , c.creator - , c.owner - , ROW_NUMBER() OVER () AS in_tx_order - FROM - UNNEST(commands) AS c -) AS c -JOIN ( - SELECT - cmds.instance_id - , cmds.aggregate_type - , cmds.aggregate_id - , CASE WHEN (e.owner IS NOT NULL OR e.owner <> '') THEN e.owner ELSE command_owners.owner END AS owner - , COALESCE(MAX(e.sequence), 0) AS sequence + , CASE WHEN c.enforce_owner + THEN c.owner + ELSE COALESCE(e.owner, c.owner) + END AS owner + , EXTRACT(EPOCH FROM NOW()) AS position + , c.in_tx_order FROM ( - SELECT DISTINCT - instance_id - , aggregate_type - , aggregate_id - , owner - FROM UNNEST(commands) - ) AS cmds - LEFT JOIN eventstore.events2 AS e - ON cmds.instance_id = e.instance_id - AND cmds.aggregate_type = e.aggregate_type - AND cmds.aggregate_id = e.aggregate_id - JOIN ( - SELECT - DISTINCT ON ( - instance_id - , aggregate_type - , aggregate_id - ) - instance_id - , aggregate_type - , aggregate_id - , owner - FROM - UNNEST(commands) - ) AS command_owners ON - cmds.instance_id = command_owners.instance_id - AND cmds.aggregate_type = command_owners.aggregate_type - AND cmds.aggregate_id = command_owners.aggregate_id - GROUP BY - cmds.instance_id - , cmds.aggregate_type - , cmds.aggregate_id - , 4 -- owner -) AS cs - ON c.instance_id = cs.instance_id - AND c.aggregate_type = cs.aggregate_type - AND c.aggregate_id = cs.aggregate_id -ORDER BY - in_tx_order; -$$ LANGUAGE SQL; + SELECT c.*, CAST(ROW_NUMBER() OVER () AS INTEGER) AS in_tx_order + FROM UNNEST(commands) AS c + ) AS c + LEFT JOIN UNNEST(_latest_events) AS e + ON c.instance_id = e.instance_id + AND c.aggregate_type = e.aggregate_type + AND c.aggregate_id = e.aggregate_id + ORDER BY + in_tx_order; +END; +$$; -CREATE OR REPLACE FUNCTION eventstore.push(commands eventstore.command[]) RETURNS SETOF eventstore.events2 VOLATILE AS $$ +CREATE OR REPLACE FUNCTION eventstore.push(commands eventstore.command2[]) RETURNS SETOF eventstore.events2 VOLATILE AS $$ INSERT INTO eventstore.events2 SELECT * FROM eventstore.commands_to_events(commands) +ORDER BY in_tx_order RETURNING * $$ LANGUAGE SQL; diff --git a/cmd/setup/40/01_type.sql b/cmd/setup/40/01_type.sql index ace6d8fe1..908aeb12c 100644 --- a/cmd/setup/40/01_type.sql +++ b/cmd/setup/40/01_type.sql @@ -1,6 +1,6 @@ -- represents an event to be created. DO $$ BEGIN - CREATE TYPE eventstore.command AS ( + CREATE TYPE eventstore.command2 AS ( instance_id TEXT , aggregate_type TEXT , aggregate_id TEXT @@ -9,6 +9,7 @@ DO $$ BEGIN , payload JSONB , creator TEXT , owner TEXT + , enforce_owner BOOLEAN ); EXCEPTION WHEN duplicate_object THEN null; diff --git a/cmd/setup/40/02_func.sql b/cmd/setup/40/02_func.sql index 47681e721..21bcf31a8 100644 --- a/cmd/setup/40/02_func.sql +++ b/cmd/setup/40/02_func.sql @@ -1,92 +1,120 @@ -CREATE OR REPLACE FUNCTION eventstore.latest_aggregate_state( - instance_id TEXT - , aggregate_type TEXT - , aggregate_id TEXT - - , sequence OUT BIGINT - , owner OUT TEXT -) - LANGUAGE 'plpgsql' - STABLE PARALLEL SAFE -AS $$ - BEGIN - SELECT - COALESCE(e.sequence, 0) AS sequence - , e.owner - INTO - sequence - , owner - FROM - eventstore.events2 e - WHERE - e.instance_id = $1 - AND e.aggregate_type = $2 - AND e.aggregate_id = $3 - ORDER BY - e.sequence DESC - LIMIT 1; +DO $$ BEGIN + CREATE TYPE eventstore.latest_command AS ( + instance_id TEXT + , aggregate_type TEXT + , aggregate_id TEXT + , owner TEXT + , sequence BIGINT + ); +EXCEPTION + WHEN duplicate_object THEN null; +END $$; - RETURN; - END; -$$; - -CREATE OR REPLACE FUNCTION eventstore.commands_to_events(commands eventstore.command[]) +CREATE OR REPLACE FUNCTION eventstore.commands_to_events(commands eventstore.command2[]) RETURNS SETOF eventstore.events2 LANGUAGE 'plpgsql' STABLE PARALLEL SAFE ROWS 10 AS $$ DECLARE - "aggregate" RECORD; - current_sequence BIGINT; - current_owner TEXT; - created_at TIMESTAMPTZ; + _latest_events eventstore.latest_command[]; + command eventstore.command2; + last_owner TEXT; + i INTEGER; BEGIN - created_at := statement_timestamp(); - FOR "aggregate" IN - SELECT DISTINCT - instance_id - , aggregate_type - , aggregate_id - FROM UNNEST(commands) - LOOP - SELECT - * - INTO - current_sequence - , current_owner - FROM eventstore.latest_aggregate_state( - "aggregate".instance_id - , "aggregate".aggregate_type - , "aggregate".aggregate_id - ); - RETURN QUERY + SELECT array_agg(res) INTO _latest_events + FROM ( SELECT - c.instance_id - , c.aggregate_type - , c.aggregate_id - , c.command_type -- AS event_type - , COALESCE(current_sequence, 0) + ROW_NUMBER() OVER () -- AS sequence - , c.revision - , created_at - , c.payload - , c.creator - , COALESCE(current_owner, c.owner) -- AS owner - , EXTRACT(EPOCH FROM created_at) -- AS position - , c.ordinality::{{ .InTxOrderType }} -- AS in_tx_order - FROM - UNNEST(commands) WITH ORDINALITY AS c - WHERE - c.instance_id = aggregate.instance_id - AND c.aggregate_type = aggregate.aggregate_type - AND c.aggregate_id = aggregate.aggregate_id; + ROW( + c.instance_id, + c.aggregate_type, + c.aggregate_id, + COALESCE(e.owner, c.owner), + COALESCE(e.sequence, 0) + )::eventstore.latest_command AS res + FROM ( + SELECT DISTINCT ON (c.instance_id, c.aggregate_type, c.aggregate_id) + c.instance_id, + c.aggregate_type, + c.aggregate_id, + c.owner, + c.ordinality + FROM ( + SELECT * + FROM UNNEST(commands) WITH ORDINALITY AS c + ) AS c + ORDER BY + c.instance_id, + c.aggregate_type, + c.aggregate_id, + c.ordinality + ) AS c + LEFT JOIN LATERAL ( + SELECT + e.owner, + e.sequence + FROM eventstore.events2 AS e + WHERE e.instance_id = c.instance_id + AND e.aggregate_type = c.aggregate_type + AND e.aggregate_id = c.aggregate_id + ORDER BY e.sequence DESC + LIMIT 1 + ) AS e ON TRUE + ) AS latest; + + FOR command IN SELECT * FROM UNNEST(commands) LOOP + IF NOT command.enforce_owner THEN + SELECT owner INTO command.owner + FROM UNNEST(_latest_events) AS e + WHERE e.instance_id = command.instance_id + AND e.aggregate_type = command.aggregate_type + AND e.aggregate_id = command.aggregate_id; + ELSE + FOR i IN 1..array_length(_latest_events, 1) LOOP + IF _latest_events[i].instance_id = command.instance_id + AND _latest_events[i].aggregate_type = command.aggregate_type + AND _latest_events[i].aggregate_id = command.aggregate_id + THEN + _latest_events[i].owner := command.owner; + EXIT; + END IF; + END LOOP; + END IF; + END LOOP; - RETURN; + + + RETURN QUERY SELECT + c.instance_id + , c.aggregate_type + , c.aggregate_id + , c.command_type AS event_type + , COALESCE(e.sequence, 0) + ROW_NUMBER() OVER (PARTITION BY c.instance_id, c.aggregate_type, c.aggregate_id ORDER BY c.in_tx_order) AS sequence + , c.revision + , NOW() AS created_at + , c.payload + , c.creator + , CASE WHEN c.enforce_owner + THEN c.owner + ELSE COALESCE(e.owner, c.owner) + END AS owner + , EXTRACT(EPOCH FROM NOW()) AS position + , c.in_tx_order + FROM ( + SELECT c.*, CAST(ROW_NUMBER() OVER () AS INTEGER) AS in_tx_order + FROM UNNEST(commands) AS c + ) AS c + LEFT JOIN UNNEST(_latest_events) AS e + ON c.instance_id = e.instance_id + AND c.aggregate_type = e.aggregate_type + AND c.aggregate_id = e.aggregate_id + ORDER BY + in_tx_order; END; $$; -CREATE OR REPLACE FUNCTION eventstore.push(commands eventstore.command[]) RETURNS SETOF eventstore.events2 VOLATILE AS $$ +CREATE OR REPLACE FUNCTION eventstore.push(commands eventstore.command2[]) RETURNS SETOF eventstore.events2 VOLATILE AS $$ INSERT INTO eventstore.events2 SELECT * FROM eventstore.commands_to_events(commands) ORDER BY in_tx_order diff --git a/cmd/setup/64.go b/cmd/setup/64.go index 906e38551..6a0e88a71 100644 --- a/cmd/setup/64.go +++ b/cmd/setup/64.go @@ -20,7 +20,7 @@ type ChangePushPosition struct { } func (mig *ChangePushPosition) Execute(ctx context.Context, _ eventstore.Event) error { - inTxOrderType, err := mig.inTxOrderType(ctx) + inTxOrderType, err := inTxOrderType(ctx, mig.dbClient) if err != nil { return err } @@ -33,8 +33,8 @@ func (mig *ChangePushPosition) String() string { return "64_change_push_position" } -func (mig *ChangePushPosition) inTxOrderType(ctx context.Context) (typeName string, err error) { - err = mig.dbClient.QueryRowContext(ctx, func(row *sql.Row) error { +func inTxOrderType(ctx context.Context, client *database.DB) (typeName string, err error) { + err = client.QueryRowContext(ctx, func(row *sql.Row) error { return row.Scan(&typeName) }, `SELECT data_type FROM information_schema.columns WHERE table_schema = 'eventstore' AND table_name = 'events2' AND column_name = 'in_tx_order'`) if err != nil { diff --git a/cmd/setup/64.sql b/cmd/setup/64.sql index 6da95fdf9..6a6a41db4 100644 --- a/cmd/setup/64.sql +++ b/cmd/setup/64.sql @@ -1,55 +1,119 @@ -CREATE OR REPLACE FUNCTION eventstore.commands_to_events(commands eventstore.command[]) +DO $$ BEGIN + CREATE TYPE eventstore.latest_command AS ( + instance_id TEXT + , aggregate_type TEXT + , aggregate_id TEXT + , owner TEXT + , sequence BIGINT + ); +EXCEPTION + WHEN duplicate_object THEN null; +END $$; + +CREATE OR REPLACE FUNCTION eventstore.commands_to_events(commands eventstore.command2[]) RETURNS SETOF eventstore.events2 LANGUAGE 'plpgsql' STABLE PARALLEL SAFE ROWS 10 AS $$ DECLARE - "aggregate" RECORD; - current_sequence BIGINT; - current_owner TEXT; - created_at TIMESTAMPTZ; + _latest_events eventstore.latest_command[]; + command eventstore.command2; + last_owner TEXT; + i INTEGER; BEGIN - created_at := statement_timestamp(); - FOR "aggregate" IN - SELECT DISTINCT - instance_id - , aggregate_type - , aggregate_id - FROM UNNEST(commands) - LOOP - SELECT - * - INTO - current_sequence - , current_owner - FROM eventstore.latest_aggregate_state( - "aggregate".instance_id - , "aggregate".aggregate_type - , "aggregate".aggregate_id - ); - RETURN QUERY + SELECT array_agg(res) INTO _latest_events + FROM ( SELECT - c.instance_id - , c.aggregate_type - , c.aggregate_id - , c.command_type -- AS event_type - , COALESCE(current_sequence, 0) + ROW_NUMBER() OVER () -- AS sequence - , c.revision - , created_at - , c.payload - , c.creator - , COALESCE(current_owner, c.owner) -- AS owner - , EXTRACT(EPOCH FROM created_at) -- AS position - , c.ordinality::%s -- AS in_tx_order - FROM - UNNEST(commands) WITH ORDINALITY AS c - WHERE - c.instance_id = aggregate.instance_id - AND c.aggregate_type = aggregate.aggregate_type - AND c.aggregate_id = aggregate.aggregate_id; + ROW( + c.instance_id, + c.aggregate_type, + c.aggregate_id, + COALESCE(e.owner, c.owner), + COALESCE(e.sequence, 0) + )::eventstore.latest_command AS res + FROM ( + SELECT DISTINCT ON (c.instance_id, c.aggregate_type, c.aggregate_id) + c.instance_id, + c.aggregate_type, + c.aggregate_id, + c.owner, + c.ordinality + FROM ( + SELECT * + FROM UNNEST(commands) WITH ORDINALITY AS c + ) AS c + ORDER BY + c.instance_id, + c.aggregate_type, + c.aggregate_id, + c.ordinality + ) AS c + LEFT JOIN LATERAL ( + SELECT + e.owner, + e.sequence + FROM eventstore.events2 AS e + WHERE e.instance_id = c.instance_id + AND e.aggregate_type = c.aggregate_type + AND e.aggregate_id = c.aggregate_id + ORDER BY e.sequence DESC + LIMIT 1 + ) AS e ON TRUE + ) AS latest; + + FOR command IN SELECT * FROM UNNEST(commands) LOOP + IF NOT command.enforce_owner THEN + SELECT owner INTO command.owner + FROM UNNEST(_latest_events) AS e + WHERE e.instance_id = command.instance_id + AND e.aggregate_type = command.aggregate_type + AND e.aggregate_id = command.aggregate_id; + ELSE + FOR i IN 1..array_length(_latest_events, 1) LOOP + IF _latest_events[i].instance_id = command.instance_id + AND _latest_events[i].aggregate_type = command.aggregate_type + AND _latest_events[i].aggregate_id = command.aggregate_id + THEN + _latest_events[i].owner := command.owner; + EXIT; + END IF; + END LOOP; + END IF; + END LOOP; - RETURN; + + + RETURN QUERY SELECT + c.instance_id + , c.aggregate_type + , c.aggregate_id + , c.command_type AS event_type + , COALESCE(e.sequence, 0) + ROW_NUMBER() OVER (PARTITION BY c.instance_id, c.aggregate_type, c.aggregate_id ORDER BY c.ordinality) AS sequence + , c.revision + , NOW() AS created_at + , c.payload + , c.creator + , CASE WHEN c.enforce_owner + THEN c.owner + ELSE COALESCE(e.owner, c.owner) + END AS owner + , EXTRACT(EPOCH FROM NOW()) AS position + , c.ordinality::%s AS in_tx_order + FROM UNNEST(commands) WITH ORDINALITY AS c + LEFT JOIN unnest(_latest_events) AS e + ON c.instance_id = e.instance_id + AND c.aggregate_type = e.aggregate_type + AND c.aggregate_id = e.aggregate_id + ORDER BY + in_tx_order; END; $$; + +CREATE OR REPLACE FUNCTION eventstore.push(commands eventstore.command2[]) RETURNS SETOF eventstore.events2 VOLATILE AS $$ +INSERT INTO eventstore.events2 +SELECT * FROM eventstore.commands_to_events(commands) +ORDER BY in_tx_order +RETURNING * +$$ LANGUAGE SQL; \ No newline at end of file diff --git a/cmd/setup/70.go b/cmd/setup/70.go new file mode 100644 index 000000000..a50b880cc --- /dev/null +++ b/cmd/setup/70.go @@ -0,0 +1,34 @@ +package setup + +import ( + "context" + _ "embed" + "fmt" + + "github.com/zitadel/zitadel/internal/database" + "github.com/zitadel/zitadel/internal/eventstore" +) + +var ( + //go:embed 70.sql + addEventStoreCommandEnforceOwnerColumn string +) + +type AddEventStoreCommandEnforceOwnerColumn struct { + dbClient *database.DB +} + +func (mig *AddEventStoreCommandEnforceOwnerColumn) Execute(ctx context.Context, _ eventstore.Event) error { + inTxOrderType, err := inTxOrderType(ctx, mig.dbClient) + if err != nil { + return err + } + + stmt := fmt.Sprintf(addEventStoreCommandEnforceOwnerColumn, inTxOrderType) + _, err = mig.dbClient.ExecContext(ctx, stmt) + return err +} + +func (mig *AddEventStoreCommandEnforceOwnerColumn) String() string { + return "70_add_eventstore_command_type_enforce_owner_column" +} diff --git a/cmd/setup/70.sql b/cmd/setup/70.sql new file mode 100644 index 000000000..81cbbc899 --- /dev/null +++ b/cmd/setup/70.sql @@ -0,0 +1,136 @@ +-- represents an event to be created. +DO $$ BEGIN + CREATE TYPE eventstore.command2 AS ( + instance_id TEXT + , aggregate_type TEXT + , aggregate_id TEXT + , command_type TEXT + , revision INT2 + , payload JSONB + , creator TEXT + , owner TEXT + , enforce_owner BOOLEAN + ); +EXCEPTION + WHEN duplicate_object THEN null; +END $$; + +DO $$ BEGIN + CREATE TYPE eventstore.latest_command AS ( + instance_id TEXT + , aggregate_type TEXT + , aggregate_id TEXT + , owner TEXT + , sequence BIGINT + ); +EXCEPTION + WHEN duplicate_object THEN null; +END $$; + +CREATE OR REPLACE FUNCTION eventstore.commands_to_events(commands eventstore.command2[]) + RETURNS SETOF eventstore.events2 + LANGUAGE 'plpgsql' + STABLE PARALLEL SAFE + ROWS 10 +AS $$ +DECLARE + _latest_events eventstore.latest_command[]; + command eventstore.command2; + last_owner TEXT; + i INTEGER; +BEGIN + + SELECT array_agg(res) INTO _latest_events + FROM ( + SELECT + ROW( + c.instance_id, + c.aggregate_type, + c.aggregate_id, + COALESCE(e.owner, c.owner), + COALESCE(e.sequence, 0) + )::eventstore.latest_command AS res + FROM ( + SELECT DISTINCT ON (c.instance_id, c.aggregate_type, c.aggregate_id) + c.instance_id, + c.aggregate_type, + c.aggregate_id, + c.owner, + c.ordinality + FROM ( + SELECT * + FROM UNNEST(commands) WITH ORDINALITY AS c + ) AS c + ORDER BY + c.instance_id, + c.aggregate_type, + c.aggregate_id, + c.ordinality + ) AS c + LEFT JOIN LATERAL ( + SELECT + e.owner, + e.sequence + FROM eventstore.events2 AS e + WHERE e.instance_id = c.instance_id + AND e.aggregate_type = c.aggregate_type + AND e.aggregate_id = c.aggregate_id + ORDER BY e.sequence DESC + LIMIT 1 + ) AS e ON TRUE + ) AS latest; + + FOR command IN SELECT * FROM UNNEST(commands) LOOP + IF NOT command.enforce_owner THEN + SELECT owner INTO command.owner + FROM UNNEST(_latest_events) AS e + WHERE e.instance_id = command.instance_id + AND e.aggregate_type = command.aggregate_type + AND e.aggregate_id = command.aggregate_id; + ELSE + FOR i IN 1..array_length(_latest_events, 1) LOOP + IF _latest_events[i].instance_id = command.instance_id + AND _latest_events[i].aggregate_type = command.aggregate_type + AND _latest_events[i].aggregate_id = command.aggregate_id + THEN + _latest_events[i].owner := command.owner; + EXIT; + END IF; + END LOOP; + END IF; + + END LOOP; + + + RETURN QUERY SELECT + c.instance_id + , c.aggregate_type + , c.aggregate_id + , c.command_type AS event_type + , COALESCE(e.sequence, 0) + ROW_NUMBER() OVER (PARTITION BY c.instance_id, c.aggregate_type, c.aggregate_id ORDER BY c.ordinality) AS sequence + , c.revision + , NOW() AS created_at + , c.payload + , c.creator + , CASE WHEN c.enforce_owner + THEN c.owner + ELSE COALESCE(e.owner, c.owner) + END AS owner + , EXTRACT(EPOCH FROM NOW()) AS position + , c.ordinality::%s AS in_tx_order + FROM UNNEST(commands) WITH ORDINALITY AS c + LEFT JOIN unnest(_latest_events) AS e + ON c.instance_id = e.instance_id + AND c.aggregate_type = e.aggregate_type + AND c.aggregate_id = e.aggregate_id + ORDER BY + in_tx_order; +END; +$$; + +CREATE OR REPLACE FUNCTION eventstore.push(commands eventstore.command2[]) RETURNS SETOF eventstore.events2 VOLATILE AS $$ +INSERT INTO eventstore.events2 +SELECT * FROM eventstore.commands_to_events(commands) +ORDER BY in_tx_order +RETURNING * +$$ LANGUAGE SQL; \ No newline at end of file diff --git a/cmd/setup/config.go b/cmd/setup/config.go index e41a94e2b..bce60cfaa 100644 --- a/cmd/setup/config.go +++ b/cmd/setup/config.go @@ -185,6 +185,7 @@ type Steps struct { s67SyncMemberRoleFields *SyncMemberRoleFields s68TargetAddPayloadTypeColumn *TargetAddPayloadTypeColumn s69CacheTablesLogged *CacheTablesLogged + s70AddEventStoreCommandEnforceOwner *AddEventStoreCommandEnforceOwnerColumn RelationalTables *TransactionalTables } diff --git a/cmd/setup/setup.go b/cmd/setup/setup.go index 3e533351f..faca769c0 100644 --- a/cmd/setup/setup.go +++ b/cmd/setup/setup.go @@ -248,6 +248,7 @@ func Setup(ctx context.Context, config *Config, steps *Steps, masterKey string) steps.s67SyncMemberRoleFields = &SyncMemberRoleFields{dbClient: dbClient} steps.s68TargetAddPayloadTypeColumn = &TargetAddPayloadTypeColumn{dbClient: dbClient} steps.s69CacheTablesLogged = &CacheTablesLogged{dbClient: dbClient} + steps.s70AddEventStoreCommandEnforceOwner = &AddEventStoreCommandEnforceOwnerColumn{dbClient: dbClient} err = projection.Create(ctx, dbClient, eventstoreClient, config.Projections, nil, nil, nil) if err != nil { @@ -257,6 +258,7 @@ func Setup(ctx context.Context, config *Config, steps *Steps, masterKey string) for _, step := range []migration.Migration{ steps.s14NewEventsTable, steps.s40InitPushFunc, + steps.s70AddEventStoreCommandEnforceOwner, steps.s1ProjectionTable, steps.s2AssetsTable, steps.s28AddFieldTable, diff --git a/internal/command/group_model.go b/internal/command/group_model.go index 5bd052acc..e8a235a0b 100644 --- a/internal/command/group_model.go +++ b/internal/command/group_model.go @@ -71,6 +71,10 @@ func (g *GroupWriteModel) Reduce() error { g.Description = *e.Description } case *group.GroupRemovedEvent: + g.Name = "" + g.Description = "" + g.UserIDs = nil + g.existingUserIDs = make(map[string]struct{}) g.State = domain.GroupStateRemoved case *group.GroupUsersAddedEvent: for _, userID := range e.UserIDs { diff --git a/internal/command/metadata_model.go b/internal/command/metadata_model.go index 94bd89470..5d78b8609 100644 --- a/internal/command/metadata_model.go +++ b/internal/command/metadata_model.go @@ -4,6 +4,8 @@ import ( "github.com/zitadel/zitadel/internal/domain" "github.com/zitadel/zitadel/internal/eventstore" "github.com/zitadel/zitadel/internal/repository/metadata" + "github.com/zitadel/zitadel/internal/repository/org" + "github.com/zitadel/zitadel/internal/repository/user" ) type MetadataWriteModel struct { @@ -17,6 +19,10 @@ type MetadataWriteModel struct { func (wm *MetadataWriteModel) Reduce() error { for _, event := range wm.Events { switch e := event.(type) { + case *user.HumanAddedEvent, *user.HumanRegisteredEvent, *user.MachineAddedEvent, *user.UserRemovedEvent, + *org.OrgAddedEvent, *org.OrgRemovedEvent: + wm.Value = nil + wm.State = domain.MetadataStateUnspecified case *metadata.SetEvent: if wm.Key != e.Key { continue @@ -44,6 +50,9 @@ type MetadataListWriteModel struct { func (wm *MetadataListWriteModel) Reduce() error { for _, event := range wm.Events { switch e := event.(type) { + case *user.HumanAddedEvent, *user.HumanRegisteredEvent, *user.MachineAddedEvent, *user.UserRemovedEvent, + *org.OrgAddedEvent, *org.OrgRemovedEvent: + wm.metadataList = make(map[string][]byte) case *metadata.SetEvent: wm.metadataList[e.Key] = e.Value case *metadata.RemovedEvent: diff --git a/internal/command/org_action_model.go b/internal/command/org_action_model.go index 6f66ee972..7b005af58 100644 --- a/internal/command/org_action_model.go +++ b/internal/command/org_action_model.go @@ -7,6 +7,7 @@ import ( "github.com/zitadel/zitadel/internal/domain" "github.com/zitadel/zitadel/internal/eventstore" "github.com/zitadel/zitadel/internal/repository/action" + "github.com/zitadel/zitadel/internal/repository/org" ) type ActionWriteModel struct { @@ -55,6 +56,10 @@ func (wm *ActionWriteModel) Reduce() error { case *action.ReactivatedEvent: wm.State = domain.ActionStateActive case *action.RemovedEvent: + wm.Name = "" + wm.Script = "" + wm.Timeout = 0 + wm.AllowedToFail = false wm.State = domain.ActionStateRemoved } } @@ -140,6 +145,8 @@ func (wm *ActionExistsModel) Reduce() error { break } } + case *org.OrgAddedEvent, *org.OrgRemovedEvent: + wm.checkedIDs = nil } } return wm.WriteModel.Reduce() @@ -151,8 +158,10 @@ func (wm *ActionExistsModel) Query() *eventstore.SearchQueryBuilder { AddQuery(). AggregateTypes(action.AggregateType). AggregateIDs(wm.actionIDs...). - EventTypes(action.AddedEventType, - action.RemovedEventType). + EventTypes(action.AddedEventType, action.RemovedEventType). + Or(). + AggregateTypes(org.AggregateType). + EventTypes(org.OrgAddedEventType, org.OrgRemovedEventType). Builder() } @@ -174,6 +183,8 @@ func NewActionsListByOrgModel(resourceOwner string) *ActionsListByOrgModel { func (wm *ActionsListByOrgModel) Reduce() error { for _, event := range wm.Events { switch e := event.(type) { + case *org.OrgAddedEvent, *org.OrgRemovedEvent: + wm.Actions = make(map[string]*ActionWriteModel) case *action.AddedEvent: wm.Actions[e.Aggregate().ID] = &ActionWriteModel{ WriteModel: eventstore.WriteModel{ @@ -198,10 +209,13 @@ func (wm *ActionsListByOrgModel) Query() *eventstore.SearchQueryBuilder { return eventstore.NewSearchQueryBuilder(eventstore.ColumnsEvent). ResourceOwner(wm.ResourceOwner). AddQuery(). - AggregateTypes(action.AggregateType). + AggregateTypes(action.AggregateType, org.AggregateType). EventTypes(action.AddedEventType, action.DeactivatedEventType, action.ReactivatedEventType, - action.RemovedEventType). + action.RemovedEventType, + org.OrgAddedEventType, + org.OrgRemovedEventType, + ). Builder() } diff --git a/internal/command/org_domain_model.go b/internal/command/org_domain_model.go index b387e6f4b..a3b057594 100644 --- a/internal/command/org_domain_model.go +++ b/internal/command/org_domain_model.go @@ -19,6 +19,14 @@ type OrgDomainWriteModel struct { State domain.OrgDomainState } +func (wm *OrgDomainWriteModel) reset() { + wm.ValidationType = domain.OrgDomainValidationTypeUnspecified + wm.ValidationCode = nil + wm.Primary = false + wm.Verified = false + wm.State = domain.OrgDomainStateUnspecified +} + func NewOrgDomainWriteModel(orgID string, domain string) *OrgDomainWriteModel { return &OrgDomainWriteModel{ WriteModel: eventstore.WriteModel{ @@ -59,6 +67,8 @@ func (wm *OrgDomainWriteModel) AppendEvents(events ...eventstore.Event) { continue } wm.WriteModel.AppendEvents(e) + case *org.OrgAddedEvent, *org.OrgRemovedEvent: + wm.WriteModel.AppendEvents(e) } } } @@ -82,6 +92,8 @@ func (wm *OrgDomainWriteModel) Reduce() error { wm.Primary = false wm.ValidationType = domain.OrgDomainValidationTypeUnspecified wm.ValidationCode = nil + case *org.OrgAddedEvent, *org.OrgRemovedEvent: + wm.reset() } } return wm.WriteModel.Reduce() @@ -99,7 +111,10 @@ func (wm *OrgDomainWriteModel) Query() *eventstore.SearchQueryBuilder { org.OrgDomainVerificationAddedEventType, org.OrgDomainVerifiedEventType, org.OrgDomainPrimarySetEventType, - org.OrgDomainRemovedEventType). + org.OrgDomainRemovedEventType, + org.OrgAddedEventType, + org.OrgRemovedEventType, + ). Builder() } @@ -132,6 +147,12 @@ func (wm *OrgDomainsWriteModel) Reduce() error { switch e := event.(type) { case *org.OrgAddedEvent: wm.OrgName = e.Name + wm.Domains = nil + wm.PrimaryDomain = "" + case *org.OrgRemovedEvent: + wm.OrgName = "" + wm.Domains = nil + wm.PrimaryDomain = "" case *org.OrgChangedEvent: wm.OrgName = e.Name case *org.DomainAddedEvent: @@ -171,7 +192,9 @@ func (wm *OrgDomainsWriteModel) Query() *eventstore.SearchQueryBuilder { org.OrgDomainVerificationAddedEventType, org.OrgDomainVerifiedEventType, org.OrgDomainPrimarySetEventType, - org.OrgDomainRemovedEventType). + org.OrgDomainRemovedEventType, + org.OrgRemovedEventType, + ). Builder() } @@ -202,7 +225,7 @@ func (wm *OrgDomainVerifiedWriteModel) AppendEvents(events ...eventstore.Event) continue } wm.WriteModel.AppendEvents(e) - case *org.OrgRemovedEvent: + case *org.OrgAddedEvent, *org.OrgRemovedEvent: wm.WriteModel.AppendEvents(e) } } @@ -216,7 +239,7 @@ func (wm *OrgDomainVerifiedWriteModel) Reduce() error { wm.ResourceOwner = e.Aggregate().ResourceOwner case *org.DomainRemovedEvent: wm.Verified = false - case *org.OrgRemovedEvent: + case *org.OrgAddedEvent, *org.OrgRemovedEvent: if wm.ResourceOwner != e.Aggregate().ID { continue } @@ -233,6 +256,8 @@ func (wm *OrgDomainVerifiedWriteModel) Query() *eventstore.SearchQueryBuilder { EventTypes( org.OrgDomainVerifiedEventType, org.OrgDomainRemovedEventType, - org.OrgRemovedEventType). + org.OrgRemovedEventType, + org.OrgAddedEventType, + ). Builder() } diff --git a/internal/command/org_metadata_model.go b/internal/command/org_metadata_model.go index 7e1038750..fea54ac58 100644 --- a/internal/command/org_metadata_model.go +++ b/internal/command/org_metadata_model.go @@ -30,6 +30,8 @@ func (wm *OrgMetadataWriteModel) AppendEvents(events ...eventstore.Event) { wm.MetadataWriteModel.AppendEvents(&e.RemovedEvent) case *org.MetadataRemovedAllEvent: wm.MetadataWriteModel.AppendEvents(&e.RemovedAllEvent) + default: + wm.MetadataWriteModel.AppendEvents(e) } } } @@ -41,6 +43,8 @@ func (wm *OrgMetadataWriteModel) Query() *eventstore.SearchQueryBuilder { AggregateIDs(wm.MetadataWriteModel.AggregateID). AggregateTypes(org.AggregateType). EventTypes( + org.OrgAddedEventType, + org.OrgRemovedEventType, org.MetadataSetType, org.MetadataRemovedType, org.MetadataRemovedAllType). @@ -72,6 +76,8 @@ func (wm *OrgMetadataListWriteModel) AppendEvents(events ...eventstore.Event) { wm.MetadataListWriteModel.AppendEvents(&e.RemovedEvent) case *org.MetadataRemovedAllEvent: wm.MetadataListWriteModel.AppendEvents(&e.RemovedAllEvent) + default: + wm.MetadataListWriteModel.AppendEvents(e) } } } @@ -87,6 +93,8 @@ func (wm *OrgMetadataListWriteModel) Query() *eventstore.SearchQueryBuilder { AggregateIDs(wm.MetadataListWriteModel.AggregateID). AggregateTypes(org.AggregateType). EventTypes( + org.OrgAddedEventType, + org.OrgRemovedEventType, org.MetadataSetType, org.MetadataRemovedType, org.MetadataRemovedAllType). diff --git a/internal/command/project_application_api_model.go b/internal/command/project_application_api_model.go index a6df8d77e..414ba04f0 100644 --- a/internal/command/project_application_api_model.go +++ b/internal/command/project_application_api_model.go @@ -123,7 +123,21 @@ func (wm *APIApplicationWriteModel) Reduce() error { case *project.APIConfigSecretHashUpdatedEvent: wm.HashedSecret = e.HashedSecret case *project.ProjectRemovedEvent: + wm.AppName = "" + wm.ClientID = "" + wm.HashedSecret = "" + wm.ClientSecretString = "" + wm.AuthMethodType = domain.APIAuthMethodTypeBasic + wm.api = false wm.State = domain.AppStateRemoved + case *project.ProjectAddedEvent: + wm.AppName = "" + wm.ClientID = "" + wm.HashedSecret = "" + wm.ClientSecretString = "" + wm.AuthMethodType = domain.APIAuthMethodTypeBasic + wm.api = false + wm.State = domain.AppStateUnspecified } } return wm.WriteModel.Reduce() @@ -149,6 +163,7 @@ func (wm *APIApplicationWriteModel) Query() *eventstore.SearchQueryBuilder { AggregateTypes(project.AggregateType). AggregateIDs(wm.AggregateID). EventTypes( + project.ProjectAddedType, project.ApplicationAddedType, project.ApplicationChangedType, project.ApplicationDeactivatedType, diff --git a/internal/command/project_application_key_model.go b/internal/command/project_application_key_model.go index 057af9ee9..deb761cea 100644 --- a/internal/command/project_application_key_model.go +++ b/internal/command/project_application_key_model.go @@ -97,7 +97,19 @@ func (wm *ApplicationKeyWriteModel) Reduce() error { case *project.ApplicationKeyRemovedEvent: wm.State = domain.AppStateRemoved case *project.ProjectRemovedEvent: + wm.ClientID = "" + wm.KeyID = "" + wm.KeyType = domain.AuthNKeyTypeNONE + wm.ExpirationDate = time.Time{} + wm.KeysAllowed = false wm.State = domain.AppStateRemoved + case *project.ProjectAddedEvent: + wm.ClientID = "" + wm.KeyID = "" + wm.KeyType = domain.AuthNKeyTypeNONE + wm.ExpirationDate = time.Time{} + wm.KeysAllowed = false + wm.State = domain.AppStateUnspecified } } return wm.WriteModel.Reduce() @@ -132,6 +144,7 @@ func (wm *ApplicationKeyWriteModel) Query() *eventstore.SearchQueryBuilder { AggregateTypes(project.AggregateType). AggregateIDs(wm.AggregateID). EventTypes( + project.ProjectAddedType, project.ApplicationRemovedType, project.OIDCConfigAddedType, project.OIDCConfigChangedType, diff --git a/internal/command/project_application_model.go b/internal/command/project_application_model.go index fd6c1dde5..c414090a6 100644 --- a/internal/command/project_application_model.go +++ b/internal/command/project_application_model.go @@ -60,7 +60,7 @@ func (wm *ApplicationWriteModel) AppendEvents(events ...eventstore.Event) { continue } wm.WriteModel.AppendEvents(e) - case *project.ProjectRemovedEvent: + default: wm.WriteModel.AppendEvents(e) } } @@ -87,7 +87,11 @@ func (wm *ApplicationWriteModel) Reduce() error { case *project.ApplicationRemovedEvent: wm.State = domain.AppStateRemoved case *project.ProjectRemovedEvent: + wm.Name = "" wm.State = domain.AppStateRemoved + case *project.ProjectAddedEvent: + wm.Name = "" + wm.State = domain.AppStateUnspecified } } return wm.WriteModel.Reduce() @@ -100,6 +104,7 @@ func (wm *ApplicationWriteModel) Query() *eventstore.SearchQueryBuilder { AggregateTypes(project.AggregateType). AggregateIDs(wm.AggregateID). EventTypes( + project.ProjectAddedType, project.ApplicationAddedType, project.ApplicationChangedType, project.ApplicationDeactivatedType, diff --git a/internal/command/project_application_oidc_model.go b/internal/command/project_application_oidc_model.go index 375bb26f5..45dcd2e8c 100644 --- a/internal/command/project_application_oidc_model.go +++ b/internal/command/project_application_oidc_model.go @@ -108,7 +108,7 @@ func (wm *OIDCApplicationWriteModel) AppendEvents(events ...eventstore.Event) { continue } wm.WriteModel.AppendEvents(e) - case *project.ProjectRemovedEvent: + default: wm.WriteModel.AppendEvents(e) } } @@ -118,6 +118,29 @@ func (wm *OIDCApplicationWriteModel) Reduce() error { for _, event := range wm.Events { switch e := event.(type) { case *project.ApplicationAddedEvent: + wm.ClientID = "" + wm.HashedSecret = "" + wm.ClientSecretString = "" + wm.RedirectUris = nil + wm.ResponseTypes = nil + wm.GrantTypes = nil + wm.ApplicationType = domain.OIDCApplicationTypeWeb + wm.AuthMethodType = domain.OIDCAuthMethodTypeBasic + wm.PostLogoutRedirectUris = nil + wm.OIDCVersion = domain.OIDCVersionV1 + wm.Compliance = nil + wm.DevMode = false + wm.AccessTokenType = domain.OIDCTokenTypeBearer + wm.AccessTokenRoleAssertion = false + wm.IDTokenRoleAssertion = false + wm.IDTokenUserinfoAssertion = false + wm.ClockSkew = 0 + wm.AdditionalOrigins = nil + wm.SkipNativeAppSuccessPage = false + wm.BackChannelLogoutURI = "" + wm.LoginVersion = domain.LoginVersionUnspecified + wm.LoginBaseURI = "" + wm.oidc = false wm.AppName = e.Name wm.State = domain.AppStateActive case *project.ApplicationChangedEvent: @@ -143,6 +166,30 @@ func (wm *OIDCApplicationWriteModel) Reduce() error { case *project.OIDCConfigSecretHashUpdatedEvent: wm.HashedSecret = e.HashedSecret case *project.ProjectRemovedEvent: + wm.AppName = "" + wm.ClientID = "" + wm.HashedSecret = "" + wm.ClientSecretString = "" + wm.RedirectUris = nil + wm.ResponseTypes = nil + wm.GrantTypes = nil + wm.ApplicationType = domain.OIDCApplicationTypeWeb + wm.AuthMethodType = domain.OIDCAuthMethodTypeBasic + wm.PostLogoutRedirectUris = nil + wm.OIDCVersion = domain.OIDCVersionV1 + wm.Compliance = nil + wm.DevMode = false + wm.AccessTokenType = domain.OIDCTokenTypeBearer + wm.AccessTokenRoleAssertion = false + wm.IDTokenRoleAssertion = false + wm.IDTokenUserinfoAssertion = false + wm.ClockSkew = 0 + wm.AdditionalOrigins = nil + wm.SkipNativeAppSuccessPage = false + wm.BackChannelLogoutURI = "" + wm.LoginVersion = domain.LoginVersionUnspecified + wm.LoginBaseURI = "" + wm.oidc = false wm.State = domain.AppStateRemoved } } @@ -237,6 +284,7 @@ func (wm *OIDCApplicationWriteModel) Query() *eventstore.SearchQueryBuilder { AggregateTypes(project.AggregateType). AggregateIDs(wm.AggregateID). EventTypes( + project.ProjectAddedType, project.ApplicationAddedType, project.ApplicationChangedType, project.ApplicationDeactivatedType, diff --git a/internal/command/project_application_saml_model.go b/internal/command/project_application_saml_model.go index f3097914f..6644131a7 100644 --- a/internal/command/project_application_saml_model.go +++ b/internal/command/project_application_saml_model.go @@ -112,7 +112,23 @@ func (wm *SAMLApplicationWriteModel) Reduce() error { case *project.SAMLConfigChangedEvent: wm.appendChangeSAMLEvent(e) case *project.ProjectRemovedEvent: + wm.AppName = "" + wm.EntityID = "" + wm.Metadata = nil + wm.MetadataURL = "" + wm.LoginVersion = domain.LoginVersionUnspecified + wm.LoginBaseURI = "" + wm.saml = false wm.State = domain.AppStateRemoved + case *project.ProjectAddedEvent: + wm.AppName = "" + wm.EntityID = "" + wm.Metadata = nil + wm.MetadataURL = "" + wm.LoginVersion = domain.LoginVersionUnspecified + wm.LoginBaseURI = "" + wm.saml = false + wm.State = domain.AppStateUnspecified } } return wm.WriteModel.Reduce() @@ -153,6 +169,7 @@ func (wm *SAMLApplicationWriteModel) Query() *eventstore.SearchQueryBuilder { AggregateTypes(project.AggregateType). AggregateIDs(wm.AggregateID). EventTypes( + project.ProjectAddedType, project.ApplicationAddedType, project.ApplicationChangedType, project.ApplicationDeactivatedType, diff --git a/internal/command/project_application_secret_model.go b/internal/command/project_application_secret_model.go index 67e431ff8..cad8b386e 100644 --- a/internal/command/project_application_secret_model.go +++ b/internal/command/project_application_secret_model.go @@ -105,7 +105,17 @@ func (wm *ApplicationSecretWriteModel) Reduce() error { case *project.APIConfigSecretHashUpdatedEvent: wm.HashedSecret = e.HashedSecret case *project.ProjectRemovedEvent: + wm.ClientID = "" + wm.HashedSecret = "" + wm.SecretAllowed = false + wm.IsAPI = false wm.State = domain.AppStateRemoved + case *project.ProjectAddedEvent: + wm.ClientID = "" + wm.HashedSecret = "" + wm.SecretAllowed = false + wm.IsAPI = false + wm.State = domain.AppStateUnspecified } } return wm.WriteModel.Reduce() @@ -144,6 +154,7 @@ func (wm *ApplicationSecretWriteModel) Query() *eventstore.SearchQueryBuilder { AggregateTypes(project.AggregateType). AggregateIDs(wm.AggregateID). EventTypes( + project.ProjectAddedType, project.ApplicationRemovedType, project.OIDCConfigAddedType, project.OIDCConfigChangedType, diff --git a/internal/command/project_grant_member_model.go b/internal/command/project_grant_member_model.go index 1d9ab0247..dcc007c86 100644 --- a/internal/command/project_grant_member_model.go +++ b/internal/command/project_grant_member_model.go @@ -75,7 +75,11 @@ func (wm *ProjectGrantMemberWriteModel) Reduce() error { case *project.GrantMemberCascadeRemovedEvent: wm.State = domain.MemberStateRemoved case *project.GrantRemovedEvent, *project.ProjectRemovedEvent: + wm.Roles = nil wm.State = domain.MemberStateRemoved + case *project.ProjectAddedEvent: + wm.Roles = nil + wm.State = domain.MemberStateUnspecified } } return wm.WriteModel.Reduce() @@ -88,6 +92,7 @@ func (wm *ProjectGrantMemberWriteModel) Query() *eventstore.SearchQueryBuilder { AggregateTypes(project.AggregateType). AggregateIDs(wm.AggregateID). EventTypes( + project.ProjectAddedType, project.GrantMemberAddedType, project.GrantMemberChangedType, project.GrantMemberRemovedType, diff --git a/internal/command/project_grant_model.go b/internal/command/project_grant_model.go index 9ef466c0d..77ee864f4 100644 --- a/internal/command/project_grant_model.go +++ b/internal/command/project_grant_model.go @@ -61,7 +61,7 @@ func (wm *ProjectGrantWriteModel) AppendEvents(events ...eventstore.Event) { wm.FoundGrantID = "" wm.WriteModel.AppendEvents(e) } - case *project.ProjectRemovedEvent: + default: wm.WriteModel.AppendEvents(e) } } @@ -104,6 +104,10 @@ func (wm *ProjectGrantWriteModel) Reduce() error { wm.State = domain.ProjectGrantStateRemoved case *project.ProjectRemovedEvent: wm.State = domain.ProjectGrantStateRemoved + wm.RoleKeys = nil + case *project.ProjectAddedEvent: + wm.State = domain.ProjectGrantStateUnspecified + wm.RoleKeys = nil } } return wm.WriteModel.Reduce() @@ -116,6 +120,7 @@ func (wm *ProjectGrantWriteModel) Query() *eventstore.SearchQueryBuilder { AggregateTypes(project.AggregateType). AggregateIDs(wm.AggregateID). EventTypes( + project.ProjectAddedType, project.GrantAddedType, project.GrantChangedType, project.GrantCascadeChangedType, diff --git a/internal/command/project_model.go b/internal/command/project_model.go index c6d25f351..bec5d4413 100644 --- a/internal/command/project_model.go +++ b/internal/command/project_model.go @@ -66,6 +66,11 @@ func (wm *ProjectWriteModel) Reduce() error { } wm.State = domain.ProjectStateActive case *project.ProjectRemovedEvent: + wm.Name = "" + wm.ProjectRoleAssertion = false + wm.ProjectRoleCheck = false + wm.HasProjectCheck = false + wm.PrivateLabelingSetting = domain.PrivateLabelingSettingUnspecified wm.State = domain.ProjectStateRemoved } } diff --git a/internal/command/project_role_model.go b/internal/command/project_role_model.go index 3fc9a6c81..b3eaa37ab 100644 --- a/internal/command/project_role_model.go +++ b/internal/command/project_role_model.go @@ -55,7 +55,7 @@ func (wm *ProjectRoleWriteModel) AppendEvents(events ...eventstore.Event) { if e.Key == wm.Key { wm.WriteModel.AppendEvents(e) } - case *project.ProjectRemovedEvent: + default: wm.WriteModel.AppendEvents(e) } } @@ -79,6 +79,8 @@ func (wm *ProjectRoleWriteModel) Reduce() error { } case *project.RoleRemovedEvent: wm.State = domain.ProjectRoleStateRemoved + case *project.ProjectAddedEvent: + wm.State = domain.ProjectRoleStateUnspecified case *project.ProjectRemovedEvent: wm.State = domain.ProjectRoleStateRemoved } @@ -93,6 +95,7 @@ func (wm *ProjectRoleWriteModel) Query() *eventstore.SearchQueryBuilder { AggregateTypes(project.AggregateType). AggregateIDs(wm.AggregateID). EventTypes( + project.ProjectAddedType, project.RoleAddedType, project.RoleChangedType, project.RoleRemovedType, diff --git a/internal/command/session_test.go b/internal/command/session_test.go index 46d952666..2ba718336 100644 --- a/internal/command/session_test.go +++ b/internal/command/session_test.go @@ -987,10 +987,6 @@ func TestCommands_updateSession(t *testing.T) { fields{ eventstore: expectEventstore( expectFilter( - eventFromEventPusher( - user.NewHumanAddedEvent(context.Background(), &user.NewAggregate("userID", "org1").Aggregate, - "username", "", "", "", "", language.English, domain.GenderUnspecified, "", false), - ), eventFromEventPusher( idpintent.NewStartedEvent(context.Background(), &idpintent.NewAggregate("id", "instance1").Aggregate, @@ -1029,7 +1025,7 @@ func TestCommands_updateSession(t *testing.T) { "userID", "org1", testNow, &language.Afrikaans), session.NewIntentCheckedEvent(context.Background(), &session.NewAggregate("sessionID", "instance1").Aggregate, testNow), - idpintent.NewConsumedEvent(context.Background(), &idpintent.NewAggregate("intent", "org1").Aggregate), + idpintent.NewConsumedEvent(context.Background(), &idpintent.NewAggregate("intent", "instance1").Aggregate), session.NewTokenSetEvent(context.Background(), &session.NewAggregate("sessionID", "instance1").Aggregate, "tokenID"), ), diff --git a/internal/command/user_grant_model.go b/internal/command/user_grant_model.go index cc1062bf7..021767012 100644 --- a/internal/command/user_grant_model.go +++ b/internal/command/user_grant_model.go @@ -134,6 +134,8 @@ func (wm *UserGrantPreConditionReadModel) Reduce() error { // We store the organization of the project for later checks, e.g. in case of a project grant wm.ProjectResourceOwner = e.Aggregate().ResourceOwner case *project.ProjectRemovedEvent: + wm.ExistingRoleKeysProject = nil + wm.ExistingRoleKeysGrant = nil wm.ProjectExists = false case *project.GrantAddedEvent: if projectGrantExistsOnOrganization(wm.ProjectGrantID, wm.ResourceOwner, e.GrantID, e.GrantedOrgID) { diff --git a/internal/command/user_human_access_token_model.go b/internal/command/user_human_access_token_model.go index 9a4c6cd2c..ab23e6d1f 100644 --- a/internal/command/user_human_access_token_model.go +++ b/internal/command/user_human_access_token_model.go @@ -82,7 +82,30 @@ func (wm *UserAccessTokenWriteModel) Reduce() error { *user.UserLockedEvent, *user.UserDeactivatedEvent, *user.UserRemovedEvent: + + wm.TokenID = "" + wm.ApplicationID = "" + wm.UserAgentID = "" + wm.Audience = nil + wm.Scopes = nil + wm.Expiration = time.Time{} + wm.PreferredLanguage = "" + wm.Reason = domain.TokenReasonUnspecified + wm.Actor = nil wm.UserState = domain.UserStateDeleted + + case *user.HumanAddedEvent, *user.HumanRegisteredEvent, *user.MachineAddedEvent: + + wm.TokenID = "" + wm.ApplicationID = "" + wm.UserAgentID = "" + wm.Audience = nil + wm.Scopes = nil + wm.Expiration = time.Time{} + wm.PreferredLanguage = "" + wm.Reason = domain.TokenReasonUnspecified + wm.Actor = nil + wm.UserState = domain.UserStateUnspecified } } return wm.WriteModel.Reduce() @@ -94,6 +117,9 @@ func (wm *UserAccessTokenWriteModel) Query() *eventstore.SearchQueryBuilder { AggregateTypes(user.AggregateType). AggregateIDs(wm.AggregateID). EventTypes( + user.HumanAddedType, + user.HumanRegisteredType, + user.MachineAddedEventType, user.UserTokenAddedType, user.UserTokenRemovedType, user.HumanSignedOutType, diff --git a/internal/command/user_human_address_model.go b/internal/command/user_human_address_model.go index 80d1faa93..baf52be94 100644 --- a/internal/command/user_human_address_model.go +++ b/internal/command/user_human_address_model.go @@ -63,7 +63,19 @@ func (wm *HumanAddressWriteModel) Reduce() error { wm.StreetAddress = *e.StreetAddress } case *user.UserRemovedEvent: + wm.Country = "" + wm.Locality = "" + wm.PostalCode = "" + wm.Region = "" + wm.StreetAddress = "" wm.State = domain.AddressStateRemoved + case *user.MachineAddedEvent: + wm.Country = "" + wm.Locality = "" + wm.PostalCode = "" + wm.Region = "" + wm.StreetAddress = "" + wm.State = domain.AddressStateUnspecified } } return wm.WriteModel.Reduce() @@ -80,6 +92,7 @@ func (wm *HumanAddressWriteModel) Query() *eventstore.SearchQueryBuilder { user.UserV1AddressChangedType, user.HumanAddedType, user.HumanRegisteredType, + user.MachineAddedEventType, user.HumanAddressChangedType, user.UserRemovedType). Builder() diff --git a/internal/command/user_human_email_model.go b/internal/command/user_human_email_model.go index b60eef5d2..6d896c38a 100644 --- a/internal/command/user_human_email_model.go +++ b/internal/command/user_human_email_model.go @@ -37,9 +37,19 @@ func (wm *HumanEmailWriteModel) Reduce() error { for _, event := range wm.Events { switch e := event.(type) { case *user.HumanAddedEvent: + wm.IsEmailVerified = false + wm.Code = nil + wm.CodeCreationDate = time.Time{} + wm.CodeExpiry = 0 + wm.AuthRequestID = "" wm.Email = e.EmailAddress wm.UserState = domain.UserStateActive case *user.HumanRegisteredEvent: + wm.IsEmailVerified = false + wm.Code = nil + wm.CodeCreationDate = time.Time{} + wm.CodeExpiry = 0 + wm.AuthRequestID = "" wm.Email = e.EmailAddress wm.UserState = domain.UserStateActive case *user.HumanInitialCodeAddedEvent: @@ -59,7 +69,23 @@ func (wm *HumanEmailWriteModel) Reduce() error { wm.IsEmailVerified = true wm.Code = nil case *user.UserRemovedEvent: + wm.Email = "" + wm.IsEmailVerified = false + wm.Code = nil + wm.CodeCreationDate = time.Time{} + wm.CodeExpiry = 0 + wm.AuthRequestID = "" + wm.UserState = domain.UserStateDeleted + case *user.MachineAddedEvent: + wm.Email = "" + wm.IsEmailVerified = false + wm.Code = nil + wm.CodeCreationDate = time.Time{} + wm.CodeExpiry = 0 + wm.AuthRequestID = "" + + wm.UserState = domain.UserStateUnspecified } } return wm.WriteModel.Reduce() @@ -74,6 +100,7 @@ func (wm *HumanEmailWriteModel) Query() *eventstore.SearchQueryBuilder { user.HumanAddedType, user.UserV1RegisteredType, user.HumanRegisteredType, + user.MachineAddedEventType, user.UserV1InitialCodeAddedType, user.HumanInitialCodeAddedType, user.UserV1InitializedCheckSucceededType, diff --git a/internal/command/user_human_init_model.go b/internal/command/user_human_init_model.go index 37c60a4b0..5db1d4793 100644 --- a/internal/command/user_human_init_model.go +++ b/internal/command/user_human_init_model.go @@ -37,9 +37,19 @@ func (wm *HumanInitCodeWriteModel) Reduce() error { for _, event := range wm.Events { switch e := event.(type) { case *user.HumanAddedEvent: + wm.IsEmailVerified = false + wm.Code = nil + wm.CodeCreationDate = time.Time{} + wm.CodeExpiry = 0 + wm.AuthRequestID = "" wm.Email = e.EmailAddress wm.UserState = domain.UserStateActive case *user.HumanRegisteredEvent: + wm.IsEmailVerified = false + wm.Code = nil + wm.CodeCreationDate = time.Time{} + wm.CodeExpiry = 0 + wm.AuthRequestID = "" wm.Email = e.EmailAddress wm.UserState = domain.UserStateActive case *user.HumanEmailChangedEvent: @@ -57,7 +67,21 @@ func (wm *HumanInitCodeWriteModel) Reduce() error { wm.Code = nil wm.UserState = domain.UserStateActive case *user.UserRemovedEvent: + wm.Email = "" + wm.IsEmailVerified = false + wm.Code = nil + wm.CodeCreationDate = time.Time{} + wm.CodeExpiry = 0 + wm.AuthRequestID = "" wm.UserState = domain.UserStateDeleted + case *user.MachineAddedEvent: + wm.Email = "" + wm.IsEmailVerified = false + wm.Code = nil + wm.CodeCreationDate = time.Time{} + wm.CodeExpiry = 0 + wm.AuthRequestID = "" + wm.UserState = domain.UserStateUnspecified } } return wm.WriteModel.Reduce() @@ -73,6 +97,7 @@ func (wm *HumanInitCodeWriteModel) Query() *eventstore.SearchQueryBuilder { user.HumanAddedType, user.UserV1RegisteredType, user.HumanRegisteredType, + user.MachineAddedEventType, user.UserV1EmailChangedType, user.HumanEmailChangedType, user.UserV1EmailVerifiedType, diff --git a/internal/command/user_human_model.go b/internal/command/user_human_model.go index 9d13d01b9..046c39744 100644 --- a/internal/command/user_human_model.go +++ b/internal/command/user_human_model.go @@ -91,7 +91,43 @@ func (wm *HumanWriteModel) Reduce() error { wm.UserState = domain.UserStateActive } case *user.UserRemovedEvent: + wm.UserName = "" + wm.FirstName = "" + wm.LastName = "" + wm.NickName = "" + wm.DisplayName = "" + wm.PreferredLanguage = language.Und + wm.Gender = domain.GenderUnspecified + wm.Avatar = "" + wm.Email = "" + wm.IsEmailVerified = false + wm.Phone = "" + wm.IsPhoneVerified = false + wm.Country = "" + wm.Locality = "" + wm.PostalCode = "" + wm.Region = "" + wm.StreetAddress = "" wm.UserState = domain.UserStateDeleted + case *user.MachineAddedEvent: + wm.UserName = "" + wm.FirstName = "" + wm.LastName = "" + wm.NickName = "" + wm.DisplayName = "" + wm.PreferredLanguage = language.Und + wm.Gender = domain.GenderUnspecified + wm.Avatar = "" + wm.Email = "" + wm.IsEmailVerified = false + wm.Phone = "" + wm.IsPhoneVerified = false + wm.Country = "" + wm.Locality = "" + wm.PostalCode = "" + wm.Region = "" + wm.StreetAddress = "" + wm.UserState = domain.UserStateUnspecified } } return wm.WriteModel.Reduce() @@ -106,6 +142,7 @@ func (wm *HumanWriteModel) Query() *eventstore.SearchQueryBuilder { EventTypes( user.HumanAddedType, user.HumanRegisteredType, + user.MachineAddedEventType, user.HumanInitialCodeAddedType, user.HumanInitializedCheckSucceededType, user.UserUserNameChangedType, @@ -143,8 +180,11 @@ func (wm *HumanWriteModel) reduceHumanAddedEvent(e *user.HumanAddedEvent) { wm.DisplayName = e.DisplayName wm.PreferredLanguage = e.PreferredLanguage wm.Gender = e.Gender + wm.Avatar = "" wm.Email = e.EmailAddress + wm.IsEmailVerified = false wm.Phone = e.PhoneNumber + wm.IsPhoneVerified = false wm.Country = e.Country wm.Locality = e.Locality wm.PostalCode = e.PostalCode @@ -161,8 +201,11 @@ func (wm *HumanWriteModel) reduceHumanRegisteredEvent(e *user.HumanRegisteredEve wm.DisplayName = e.DisplayName wm.PreferredLanguage = e.PreferredLanguage wm.Gender = e.Gender + wm.Avatar = "" wm.Email = e.EmailAddress + wm.IsEmailVerified = false wm.Phone = e.PhoneNumber + wm.IsPhoneVerified = false wm.Country = e.Country wm.Locality = e.Locality wm.PostalCode = e.PostalCode diff --git a/internal/command/user_human_otp_model.go b/internal/command/user_human_otp_model.go index 7119adfea..6bbc47110 100644 --- a/internal/command/user_human_otp_model.go +++ b/internal/command/user_human_otp_model.go @@ -48,7 +48,15 @@ func (wm *HumanTOTPWriteModel) Reduce() error { case *user.HumanOTPRemovedEvent: wm.State = domain.MFAStateRemoved case *user.UserRemovedEvent: + wm.Secret = nil + wm.CheckFailedCount = 0 + wm.UserLocked = false wm.State = domain.MFAStateRemoved + case *user.HumanAddedEvent, *user.HumanRegisteredEvent, *user.MachineAddedEvent: + wm.Secret = nil + wm.CheckFailedCount = 0 + wm.UserLocked = false + wm.State = domain.MFAStateUnspecified } } return wm.WriteModel.Reduce() @@ -59,7 +67,11 @@ func (wm *HumanTOTPWriteModel) Query() *eventstore.SearchQueryBuilder { AddQuery(). AggregateTypes(user.AggregateType). AggregateIDs(wm.AggregateID). - EventTypes(user.HumanMFAOTPAddedType, + EventTypes( + user.HumanAddedType, + user.HumanRegisteredType, + user.MachineAddedEventType, + user.HumanMFAOTPAddedType, user.HumanMFAOTPVerifiedType, user.HumanMFAOTPRemovedType, user.HumanMFAOTPCheckSucceededType, @@ -129,7 +141,11 @@ func (wm *HumanOTPSMSWriteModel) Reduce() error { case *user.HumanOTPSMSRemovedEvent: wm.otpAdded = false case *user.HumanPhoneRemovedEvent, - *user.UserRemovedEvent: + *user.UserRemovedEvent, + *user.HumanAddedEvent, + *user.HumanRegisteredEvent, + *user.MachineAddedEvent: + wm.phoneVerified = false wm.otpAdded = false } @@ -147,6 +163,9 @@ func (wm *HumanOTPSMSWriteModel) Query() *eventstore.SearchQueryBuilder { user.HumanOTPSMSRemovedType, user.HumanPhoneRemovedType, user.UserRemovedType, + user.HumanAddedType, + user.HumanRegisteredType, + user.MachineAddedEventType, ). Builder() @@ -300,7 +319,7 @@ func (wm *HumanOTPEmailWriteModel) Reduce() error { wm.otpAdded = true case *user.HumanOTPEmailRemovedEvent: wm.otpAdded = false - case *user.UserRemovedEvent: + case *user.UserRemovedEvent, *user.HumanAddedEvent, *user.HumanRegisteredEvent, *user.MachineAddedEvent: wm.emailVerified = false wm.otpAdded = false } @@ -314,6 +333,9 @@ func (wm *HumanOTPEmailWriteModel) Query() *eventstore.SearchQueryBuilder { AggregateTypes(user.AggregateType). AggregateIDs(wm.AggregateID). EventTypes( + user.HumanAddedType, + user.HumanRegisteredType, + user.MachineAddedEventType, user.HumanEmailVerifiedType, user.HumanOTPEmailAddedType, user.HumanOTPEmailRemovedType, diff --git a/internal/command/user_human_password_model.go b/internal/command/user_human_password_model.go index e08552080..1518a0c48 100644 --- a/internal/command/user_human_password_model.go +++ b/internal/command/user_human_password_model.go @@ -58,13 +58,35 @@ func (wm *HumanPasswordWriteModel) Reduce() error { for _, event := range wm.Events { switch e := event.(type) { case *user.HumanAddedEvent: + wm.Code = nil + wm.CodeCreationDate = time.Time{} + wm.CodeExpiry = 0 + wm.PasswordCheckFailedCount = 0 + wm.GeneratorID = "" + wm.VerificationID = "" wm.EncodedHash = crypto.SecretOrEncodedHash(e.Secret, e.EncodedHash) wm.SecretChangeRequired = e.ChangeRequired wm.UserState = domain.UserStateActive case *user.HumanRegisteredEvent: + wm.Code = nil + wm.CodeCreationDate = time.Time{} + wm.CodeExpiry = 0 + wm.PasswordCheckFailedCount = 0 + wm.GeneratorID = "" + wm.VerificationID = "" wm.EncodedHash = crypto.SecretOrEncodedHash(e.Secret, e.EncodedHash) wm.SecretChangeRequired = e.ChangeRequired wm.UserState = domain.UserStateActive + case *user.MachineAddedEvent: + wm.Code = nil + wm.CodeCreationDate = time.Time{} + wm.CodeExpiry = 0 + wm.PasswordCheckFailedCount = 0 + wm.GeneratorID = "" + wm.VerificationID = "" + wm.EncodedHash = "" + wm.SecretChangeRequired = false + wm.UserState = domain.UserStateUnspecified case *user.HumanInitialCodeAddedEvent: wm.UserState = domain.UserStateInitial case *user.HumanInitializedCheckSucceededEvent: @@ -98,6 +120,14 @@ func (wm *HumanPasswordWriteModel) Reduce() error { wm.UserState = domain.UserStateActive } case *user.UserRemovedEvent: + wm.EncodedHash = "" + wm.SecretChangeRequired = false + wm.Code = nil + wm.CodeCreationDate = time.Time{} + wm.CodeExpiry = 0 + wm.PasswordCheckFailedCount = 0 + wm.GeneratorID = "" + wm.VerificationID = "" wm.UserState = domain.UserStateDeleted case *user.HumanPasswordHashUpdatedEvent: wm.EncodedHash = e.EncodedHash @@ -113,6 +143,7 @@ func (wm *HumanPasswordWriteModel) Query() *eventstore.SearchQueryBuilder { AggregateIDs(wm.AggregateID). EventTypes(user.HumanAddedType, user.HumanRegisteredType, + user.MachineAddedEventType, user.HumanInitialCodeAddedType, user.HumanInitializedCheckSucceededType, user.HumanPasswordChangedType, diff --git a/internal/command/user_human_phone_model.go b/internal/command/user_human_phone_model.go index 41e8e8457..aab82f587 100644 --- a/internal/command/user_human_phone_model.go +++ b/internal/command/user_human_phone_model.go @@ -39,17 +39,45 @@ func (wm *HumanPhoneWriteModel) Reduce() error { for _, event := range wm.Events { switch e := event.(type) { case *user.HumanAddedEvent: - if e.PhoneNumber != "" { - wm.Phone = e.PhoneNumber + wm.IsPhoneVerified = false + wm.Code = nil + wm.CodeCreationDate = time.Time{} + wm.CodeExpiry = 0 + wm.GeneratorID = "" + wm.VerificationID = "" + wm.Phone = e.PhoneNumber + + if wm.Phone != "" { wm.State = domain.PhoneStateActive + } else { + wm.State = domain.PhoneStateUnspecified } wm.UserState = domain.UserStateActive case *user.HumanRegisteredEvent: - if e.PhoneNumber != "" { - wm.Phone = e.PhoneNumber + wm.IsPhoneVerified = false + wm.Code = nil + wm.CodeCreationDate = time.Time{} + wm.CodeExpiry = 0 + wm.GeneratorID = "" + wm.VerificationID = "" + wm.Phone = e.PhoneNumber + + if wm.Phone != "" { wm.State = domain.PhoneStateActive + } else { + wm.State = domain.PhoneStateUnspecified } wm.UserState = domain.UserStateActive + case *user.MachineAddedEvent: + wm.IsPhoneVerified = false + wm.Code = nil + wm.CodeCreationDate = time.Time{} + wm.CodeExpiry = 0 + wm.GeneratorID = "" + wm.VerificationID = "" + wm.Phone = "" + wm.State = domain.PhoneStateUnspecified + wm.UserState = domain.UserStateUnspecified case *user.HumanInitialCodeAddedEvent: wm.UserState = domain.UserStateInitial case *user.HumanInitializedCheckSucceededEvent: @@ -75,6 +103,12 @@ func (wm *HumanPhoneWriteModel) Reduce() error { wm.IsPhoneVerified = false wm.Phone = "" case *user.UserRemovedEvent: + wm.IsPhoneVerified = false + wm.Code = nil + wm.CodeCreationDate = time.Time{} + wm.CodeExpiry = 0 + wm.GeneratorID = "" + wm.VerificationID = "" wm.UserState = domain.UserStateDeleted wm.IsPhoneVerified = false wm.Phone = "" diff --git a/internal/command/user_human_profile_model.go b/internal/command/user_human_profile_model.go index 608ac13ca..dbbd8a8e9 100644 --- a/internal/command/user_human_profile_model.go +++ b/internal/command/user_human_profile_model.go @@ -71,7 +71,21 @@ func (wm *HumanProfileWriteModel) Reduce() error { wm.Gender = *e.Gender } case *user.UserRemovedEvent: + wm.FirstName = "" + wm.LastName = "" + wm.NickName = "" + wm.DisplayName = "" + wm.PreferredLanguage = language.Und + wm.Gender = domain.GenderUnspecified wm.UserState = domain.UserStateDeleted + case *user.MachineAddedEvent: + wm.FirstName = "" + wm.LastName = "" + wm.NickName = "" + wm.DisplayName = "" + wm.PreferredLanguage = language.Und + wm.Gender = domain.GenderUnspecified + wm.UserState = domain.UserStateUnspecified } } return wm.WriteModel.Reduce() @@ -83,8 +97,10 @@ func (wm *HumanProfileWriteModel) Query() *eventstore.SearchQueryBuilder { AddQuery(). AggregateTypes(user.AggregateType). AggregateIDs(wm.AggregateID). - EventTypes(user.HumanAddedType, + EventTypes( + user.HumanAddedType, user.HumanRegisteredType, + user.MachineAddedEventType, user.HumanProfileChangedType, user.UserRemovedType, user.UserV1AddedType, diff --git a/internal/command/user_human_recovery_code_model.go b/internal/command/user_human_recovery_code_model.go index a56951756..a8b3bd1b5 100644 --- a/internal/command/user_human_recovery_code_model.go +++ b/internal/command/user_human_recovery_code_model.go @@ -63,7 +63,15 @@ func (wm *HumanRecoveryCodeWriteModel) Reduce() error { wm.userLocked = false wm.FailedAttempts = 0 case *user.UserRemovedEvent: + wm.FailedAttempts = 0 + wm.codes = nil + wm.userLocked = false wm.State = domain.MFAStateRemoved + case *user.HumanAddedEvent, *user.HumanRegisteredEvent, *user.MachineAddedEvent: + wm.FailedAttempts = 0 + wm.codes = nil + wm.userLocked = false + wm.State = domain.MFAStateUnspecified } } return wm.WriteModel.Reduce() @@ -74,7 +82,11 @@ func (wm *HumanRecoveryCodeWriteModel) Query() *eventstore.SearchQueryBuilder { AddQuery(). AggregateTypes(user.AggregateType). AggregateIDs(wm.AggregateID). - EventTypes(user.HumanRecoveryCodesAddedType, + EventTypes( + user.HumanAddedType, + user.HumanRegisteredType, + user.MachineAddedEventType, + user.HumanRecoveryCodesAddedType, user.HumanRecoveryCodesRemovedType, user.HumanRecoveryCodeCheckSucceededType, user.HumanRecoveryCodeCheckFailedType, diff --git a/internal/command/user_human_recovery_code_model_test.go b/internal/command/user_human_recovery_code_model_test.go index 7f7b4d83e..923f177a7 100644 --- a/internal/command/user_human_recovery_code_model_test.go +++ b/internal/command/user_human_recovery_code_model_test.go @@ -309,7 +309,7 @@ func TestHumanRecoveryCodeWriteModel_Reduce(t *testing.T) { }, State: domain.MFAStateRemoved, FailedAttempts: 0, - codes: []string{"code1", "code2", "code3"}, + codes: nil, userLocked: false, }, }, diff --git a/internal/command/user_human_refresh_token_model.go b/internal/command/user_human_refresh_token_model.go index 45ba54e73..cf2a672a2 100644 --- a/internal/command/user_human_refresh_token_model.go +++ b/internal/command/user_human_refresh_token_model.go @@ -84,7 +84,25 @@ func (wm *HumanRefreshTokenWriteModel) Reduce() error { *user.UserLockedEvent, *user.UserDeactivatedEvent, *user.UserRemovedEvent: + + wm.TokenID = "" + wm.RefreshToken = "" + wm.AuthTime = time.Time{} + wm.IdleExpiration = time.Time{} + wm.Expiration = time.Time{} + wm.UserAgentID = "" + wm.AuthMethodsReferences = nil + wm.Actor = nil wm.UserState = domain.UserStateDeleted + case *user.HumanAddedEvent, *user.HumanRegisteredEvent, *user.MachineAddedEvent: + wm.TokenID = "" + wm.RefreshToken = "" + wm.AuthTime = time.Time{} + wm.IdleExpiration = time.Time{} + wm.Expiration = time.Time{} + wm.UserAgentID = "" + wm.AuthMethodsReferences = nil + wm.Actor = nil } } return wm.WriteModel.Reduce() @@ -96,6 +114,9 @@ func (wm *HumanRefreshTokenWriteModel) Query() *eventstore.SearchQueryBuilder { AggregateTypes(user.AggregateType). AggregateIDs(wm.AggregateID). EventTypes( + user.HumanAddedType, + user.HumanRegisteredType, + user.MachineAddedEventType, user.HumanRefreshTokenAddedType, user.HumanRefreshTokenRenewedType, user.HumanRefreshTokenRemovedType, diff --git a/internal/command/user_human_webauthn_model.go b/internal/command/user_human_webauthn_model.go index d2fa39a59..58cd0b35a 100644 --- a/internal/command/user_human_webauthn_model.go +++ b/internal/command/user_human_webauthn_model.go @@ -87,7 +87,7 @@ func (wm *HumanWebAuthNWriteModel) AppendEvents(events ...eventstore.Event) { if wm.WebauthNTokenID == e.WebAuthNTokenID { wm.WriteModel.AppendEvents(&e.HumanWebAuthNRemovedEvent) } - case *user.UserRemovedEvent: + default: wm.WriteModel.AppendEvents(e) } } @@ -104,7 +104,15 @@ func (wm *HumanWebAuthNWriteModel) Reduce() error { wm.SignCount = e.SignCount case *user.HumanWebAuthNRemovedEvent: wm.State = domain.MFAStateRemoved - case *user.UserRemovedEvent: + case *user.HumanAddedEvent, *user.HumanRegisteredEvent, *user.MachineAddedEvent, *user.UserRemovedEvent: + wm.Challenge = "" + wm.KeyID = nil + wm.PublicKey = nil + wm.AttestationType = "" + wm.AAGUID = nil + wm.SignCount = 0 + wm.WebAuthNTokenName = "" + wm.RPID = "" wm.State = domain.MFAStateRemoved } } @@ -134,7 +142,11 @@ func (wm *HumanWebAuthNWriteModel) Query() *eventstore.SearchQueryBuilder { AddQuery(). AggregateTypes(user.AggregateType). AggregateIDs(wm.AggregateID). - EventTypes(user.HumanU2FTokenAddedType, + EventTypes( + user.HumanAddedType, + user.HumanRegisteredType, + user.MachineAddedEventType, + user.HumanU2FTokenAddedType, user.HumanPasswordlessTokenAddedType, user.HumanU2FTokenAddedType, user.HumanPasswordlessTokenAddedType, @@ -206,7 +218,11 @@ func (wm *HumanU2FTokensReadModel) Reduce() error { wm.WebAuthNTokens[len(wm.WebAuthNTokens)-1] = nil wm.WebAuthNTokens = wm.WebAuthNTokens[:len(wm.WebAuthNTokens)-1] case *user.UserRemovedEvent: + wm.WebAuthNTokens = nil wm.UserState = domain.UserStateDeleted + case *user.HumanAddedEvent, *user.HumanRegisteredEvent, *user.MachineAddedEvent: + wm.WebAuthNTokens = nil + wm.UserState = domain.UserStateUnspecified } } return wm.WriteModel.Reduce() @@ -221,7 +237,12 @@ func (rm *HumanU2FTokensReadModel) Query() *eventstore.SearchQueryBuilder { EventTypes( user.HumanU2FTokenAddedType, user.HumanU2FTokenVerifiedType, - user.HumanU2FTokenRemovedType). + user.HumanU2FTokenRemovedType, + user.UserRemovedType, + user.HumanAddedType, + user.HumanRegisteredType, + user.MachineAddedEventType, + ). Builder() } @@ -293,7 +314,11 @@ func (wm *HumanPasswordlessTokensReadModel) Reduce() error { wm.WebAuthNTokens[len(wm.WebAuthNTokens)-1] = nil wm.WebAuthNTokens = wm.WebAuthNTokens[:len(wm.WebAuthNTokens)-1] case *user.UserRemovedEvent: + wm.WebAuthNTokens = nil wm.UserState = domain.UserStateDeleted + case *user.HumanAddedEvent, *user.HumanRegisteredEvent, *user.MachineAddedEvent: + wm.WebAuthNTokens = nil + wm.UserState = domain.UserStateUnspecified } } return wm.WriteModel.Reduce() @@ -308,7 +333,12 @@ func (rm *HumanPasswordlessTokensReadModel) Query() *eventstore.SearchQueryBuild EventTypes( user.HumanPasswordlessTokenAddedType, user.HumanPasswordlessTokenVerifiedType, - user.HumanPasswordlessTokenRemovedType). + user.HumanPasswordlessTokenRemovedType, + user.UserRemovedType, + user.HumanAddedType, + user.HumanRegisteredType, + user.MachineAddedEventType, + ). Builder() } @@ -354,7 +384,7 @@ func (wm *HumanU2FLoginReadModel) AppendEvents(events ...eventstore.Event) { continue } wm.WriteModel.AppendEvents(e) - case *user.UserRemovedEvent: + default: wm.WriteModel.AppendEvents(e) } } @@ -369,7 +399,15 @@ func (wm *HumanU2FLoginReadModel) Reduce() error { wm.UserVerification = e.UserVerification wm.State = domain.UserStateActive case *user.UserRemovedEvent: + wm.Challenge = "" + wm.AllowedCredentialIDs = nil + wm.UserVerification = domain.UserVerificationRequirementUnspecified wm.State = domain.UserStateDeleted + case *user.HumanAddedEvent, *user.HumanRegisteredEvent, *user.MachineAddedEvent: + wm.Challenge = "" + wm.AllowedCredentialIDs = nil + wm.UserVerification = domain.UserVerificationRequirementUnspecified + wm.State = domain.UserStateUnspecified } } return wm.WriteModel.Reduce() @@ -383,7 +421,11 @@ func (rm *HumanU2FLoginReadModel) Query() *eventstore.SearchQueryBuilder { AggregateIDs(rm.AggregateID). EventTypes( user.HumanU2FTokenBeginLoginType, - user.UserRemovedType). + user.UserRemovedType, + user.HumanAddedType, + user.HumanRegisteredType, + user.MachineAddedEventType, + ). Builder() } @@ -415,7 +457,7 @@ func (wm *HumanPasswordlessLoginReadModel) AppendEvents(events ...eventstore.Eve continue } wm.WriteModel.AppendEvents(e) - case *user.UserRemovedEvent: + default: wm.WriteModel.AppendEvents(e) } } @@ -430,7 +472,15 @@ func (wm *HumanPasswordlessLoginReadModel) Reduce() error { wm.UserVerification = e.UserVerification wm.State = domain.UserStateActive case *user.UserRemovedEvent: + wm.Challenge = "" + wm.AllowedCredentialIDs = nil + wm.UserVerification = domain.UserVerificationRequirementUnspecified wm.State = domain.UserStateDeleted + case *user.HumanAddedEvent, *user.HumanRegisteredEvent, *user.MachineAddedEvent: + wm.Challenge = "" + wm.AllowedCredentialIDs = nil + wm.UserVerification = domain.UserVerificationRequirementUnspecified + wm.State = domain.UserStateUnspecified } } return wm.WriteModel.Reduce() @@ -444,7 +494,11 @@ func (rm *HumanPasswordlessLoginReadModel) Query() *eventstore.SearchQueryBuilde AggregateIDs(rm.AggregateID). EventTypes( user.HumanPasswordlessTokenBeginLoginType, - user.UserRemovedType). + user.UserRemovedType, + user.HumanAddedType, + user.HumanRegisteredType, + user.MachineAddedEventType, + ). Builder() } @@ -493,7 +547,7 @@ func (wm *HumanPasswordlessInitCodeWriteModel) AppendEvents(events ...eventstore if wm.CodeID == e.ID { wm.WriteModel.AppendEvents(e) } - case *user.UserRemovedEvent: + default: wm.WriteModel.AppendEvents(e) } } @@ -512,7 +566,12 @@ func (wm *HumanPasswordlessInitCodeWriteModel) Reduce() error { wm.appendCheckFailedEvent(e) case *user.HumanPasswordlessInitCodeCheckSucceededEvent: wm.State = domain.PasswordlessInitCodeStateRemoved - case *user.UserRemovedEvent: + case *user.HumanAddedEvent, *user.HumanRegisteredEvent, *user.MachineAddedEvent, *user.UserRemovedEvent: + wm.Attempts = 0 + wm.CryptoCode = nil + wm.CodeCreationDate = time.Time{} + wm.CodeExpiration = 0 + wm.State = domain.PasswordlessInitCodeStateUnspecified wm.State = domain.PasswordlessInitCodeStateRemoved } } @@ -554,6 +613,10 @@ func (wm *HumanPasswordlessInitCodeWriteModel) Query() *eventstore.SearchQueryBu user.HumanPasswordlessInitCodeSentType, user.HumanPasswordlessInitCodeCheckFailedType, user.HumanPasswordlessInitCodeCheckSucceededType, - user.UserRemovedType). + user.UserRemovedType, + user.HumanAddedType, + user.HumanRegisteredType, + user.MachineAddedEventType, + ). Builder() } diff --git a/internal/command/user_idp_link_model.go b/internal/command/user_idp_link_model.go index edcd1ebd2..11aa2e592 100644 --- a/internal/command/user_idp_link_model.go +++ b/internal/command/user_idp_link_model.go @@ -50,7 +50,7 @@ func (wm *UserIDPLinkWriteModel) AppendEvents(events ...eventstore.Event) { continue } wm.WriteModel.AppendEvents(e) - case *user.UserRemovedEvent: + default: wm.WriteModel.AppendEvents(e) } } @@ -70,7 +70,11 @@ func (wm *UserIDPLinkWriteModel) Reduce() error { wm.State = domain.UserIDPLinkStateRemoved case *user.UserIDPLinkCascadeRemovedEvent: wm.State = domain.UserIDPLinkStateRemoved + case *user.HumanAddedEvent, *user.HumanRegisteredEvent, *user.MachineAddedEvent: + wm.DisplayName = "" + wm.State = domain.UserIDPLinkStateUnspecified case *user.UserRemovedEvent: + wm.DisplayName = "" wm.State = domain.UserIDPLinkStateRemoved } } @@ -83,7 +87,11 @@ func (wm *UserIDPLinkWriteModel) Query() *eventstore.SearchQueryBuilder { AddQuery(). AggregateTypes(user.AggregateType). AggregateIDs(wm.AggregateID). - EventTypes(user.UserIDPLinkAddedType, + EventTypes( + user.HumanAddedType, + user.HumanRegisteredType, + user.MachineAddedEventType, + user.UserIDPLinkAddedType, user.UserIDPExternalIDMigratedType, user.UserIDPLinkRemovedType, user.UserIDPLinkCascadeRemovedType, diff --git a/internal/command/user_machine_key_model.go b/internal/command/user_machine_key_model.go index d9ba62865..ac006bdd7 100644 --- a/internal/command/user_machine_key_model.go +++ b/internal/command/user_machine_key_model.go @@ -41,7 +41,7 @@ func (wm *MachineKeyWriteModel) AppendEvents(events ...eventstore.Event) { continue } wm.WriteModel.AppendEvents(e) - case *user.UserRemovedEvent: + default: wm.WriteModel.AppendEvents(e) } } @@ -50,6 +50,14 @@ func (wm *MachineKeyWriteModel) AppendEvents(events ...eventstore.Event) { func (wm *MachineKeyWriteModel) Reduce() error { for _, event := range wm.Events { switch e := event.(type) { + case *user.HumanAddedEvent, *user.HumanRegisteredEvent: + wm.KeyType = domain.AuthNKeyTypeNONE + wm.ExpirationDate = time.Time{} + wm.State = domain.MachineKeyStateUnspecified + case *user.MachineAddedEvent: + wm.KeyType = domain.AuthNKeyTypeNONE + wm.ExpirationDate = time.Time{} + wm.State = domain.MachineKeyStateUnspecified case *user.MachineKeyAddedEvent: wm.KeyID = e.KeyID wm.KeyType = e.KeyType @@ -58,6 +66,8 @@ func (wm *MachineKeyWriteModel) Reduce() error { case *user.MachineKeyRemovedEvent: wm.State = domain.MachineKeyStateRemoved case *user.UserRemovedEvent: + wm.KeyType = domain.AuthNKeyTypeNONE + wm.ExpirationDate = time.Time{} wm.State = domain.MachineKeyStateRemoved } } @@ -71,6 +81,9 @@ func (wm *MachineKeyWriteModel) Query() *eventstore.SearchQueryBuilder { AggregateTypes(user.AggregateType). AggregateIDs(wm.AggregateID). EventTypes( + user.HumanAddedType, + user.HumanRegisteredType, + user.MachineAddedEventType, user.MachineKeyAddedEventType, user.MachineKeyRemovedEventType, user.UserRemovedType). diff --git a/internal/command/user_machine_model.go b/internal/command/user_machine_model.go index e905eba4d..f6ae222ef 100644 --- a/internal/command/user_machine_model.go +++ b/internal/command/user_machine_model.go @@ -34,11 +34,19 @@ func (wm *MachineWriteModel) Reduce() error { for _, event := range wm.Events { switch e := event.(type) { case *user.MachineAddedEvent: + wm.HashedSecret = "" wm.UserName = e.UserName wm.Name = e.Name wm.Description = e.Description wm.AccessTokenType = e.AccessTokenType wm.UserState = domain.UserStateActive + case *user.HumanAddedEvent, *user.HumanRegisteredEvent: + wm.Name = "" + wm.Description = "" + wm.AccessTokenType = domain.OIDCTokenTypeBearer + wm.HashedSecret = "" + // We set the state to unspecified because it is not a valid machine. + wm.UserState = domain.UserStateUnspecified case *user.UsernameChangedEvent: wm.UserName = e.UserName case *user.MachineChangedEvent: @@ -68,6 +76,10 @@ func (wm *MachineWriteModel) Reduce() error { wm.UserState = domain.UserStateActive } case *user.UserRemovedEvent: + wm.Name = "" + wm.Description = "" + wm.AccessTokenType = domain.OIDCTokenTypeBearer + wm.HashedSecret = "" wm.UserState = domain.UserStateDeleted case *user.MachineSecretSetEvent: wm.HashedSecret = crypto.SecretOrEncodedHash(e.ClientSecret, e.HashedSecret) @@ -86,7 +98,10 @@ func (wm *MachineWriteModel) Query() *eventstore.SearchQueryBuilder { AddQuery(). AggregateTypes(user.AggregateType). AggregateIDs(wm.AggregateID). - EventTypes(user.MachineAddedEventType, + EventTypes( + user.MachineAddedEventType, + user.HumanAddedType, + user.HumanRegisteredType, user.UserUserNameChangedType, user.MachineChangedEventType, user.UserLockedType, diff --git a/internal/command/user_metadata_model.go b/internal/command/user_metadata_model.go index 3ab68242e..37bc5385d 100644 --- a/internal/command/user_metadata_model.go +++ b/internal/command/user_metadata_model.go @@ -31,6 +31,8 @@ func (wm *UserMetadataWriteModel) AppendEvents(events ...eventstore.Event) { wm.MetadataWriteModel.AppendEvents(&e.RemovedEvent) case *user.MetadataRemovedAllEvent: wm.MetadataWriteModel.AppendEvents(&e.RemovedAllEvent) + default: + wm.MetadataWriteModel.AppendEvents(e) } } } @@ -42,6 +44,10 @@ func (wm *UserMetadataWriteModel) Query() *eventstore.SearchQueryBuilder { AggregateIDs(wm.MetadataWriteModel.AggregateID). AggregateTypes(user.AggregateType). EventTypes( + user.HumanAddedType, + user.HumanRegisteredType, + user.MachineAddedEventType, + user.UserRemovedType, user.MetadataSetType, user.MetadataRemovedType, user.MetadataRemovedAllType). @@ -73,6 +79,8 @@ func (wm *UserMetadataListWriteModel) AppendEvents(events ...eventstore.Event) { wm.MetadataListWriteModel.AppendEvents(&e.RemovedEvent) case *user.MetadataRemovedAllEvent: wm.MetadataListWriteModel.AppendEvents(&e.RemovedAllEvent) + default: + wm.MetadataListWriteModel.AppendEvents(e) } } } @@ -88,6 +96,10 @@ func (wm *UserMetadataListWriteModel) Query() *eventstore.SearchQueryBuilder { AggregateIDs(wm.MetadataListWriteModel.AggregateID). AggregateTypes(user.AggregateType). EventTypes( + user.HumanAddedType, + user.HumanRegisteredType, + user.MachineAddedEventType, + user.UserRemovedType, user.MetadataSetType, user.MetadataRemovedType, user.MetadataRemovedAllType). @@ -117,6 +129,8 @@ func (wm *UserMetadataByOrgListWriteModel) AppendEvents(events ...eventstore.Eve wm.WriteModel.AppendEvents(&e.RemovedEvent) case *user.MetadataRemovedAllEvent: wm.WriteModel.AppendEvents(&e.RemovedAllEvent) + default: + wm.WriteModel.AppendEvents(e) } } } @@ -136,7 +150,7 @@ func (wm *UserMetadataByOrgListWriteModel) Reduce() error { if val, ok := wm.UserMetadata[e.Aggregate().ID]; ok { delete(val, e.Key) } - case *metadata.RemovedAllEvent: + case *metadata.RemovedAllEvent, *user.HumanAddedEvent, *user.HumanRegisteredEvent, *user.MachineAddedEvent, *user.UserRemovedEvent: delete(wm.UserMetadata, e.Aggregate().ID) } } @@ -149,6 +163,10 @@ func (wm *UserMetadataByOrgListWriteModel) Query() *eventstore.SearchQueryBuilde AddQuery(). AggregateTypes(user.AggregateType). EventTypes( + user.HumanAddedType, + user.HumanRegisteredType, + user.MachineAddedEventType, + user.UserRemovedType, user.MetadataSetType, user.MetadataRemovedType, user.MetadataRemovedAllType). diff --git a/internal/command/user_model.go b/internal/command/user_model.go index d600f9d98..5d8498d31 100644 --- a/internal/command/user_model.go +++ b/internal/command/user_model.go @@ -34,10 +34,12 @@ func (wm *UserWriteModel) Reduce() error { wm.UserName = e.UserName wm.UserState = domain.UserStateActive wm.UserType = domain.UserTypeHuman + wm.IDPLinks = nil case *user.HumanRegisteredEvent: wm.UserName = e.UserName wm.UserState = domain.UserStateActive wm.UserType = domain.UserTypeHuman + wm.IDPLinks = nil case *user.HumanInitialCodeAddedEvent: wm.UserState = domain.UserStateInitial case *user.HumanInitializedCheckSucceededEvent: @@ -64,6 +66,7 @@ func (wm *UserWriteModel) Reduce() error { wm.UserName = e.UserName wm.UserState = domain.UserStateActive wm.UserType = domain.UserTypeMachine + wm.IDPLinks = nil case *user.UsernameChangedEvent: wm.UserName = e.UserName case *user.UserLockedEvent: @@ -83,6 +86,9 @@ func (wm *UserWriteModel) Reduce() error { wm.UserState = domain.UserStateActive } case *user.UserRemovedEvent: + wm.UserName = "" + wm.IDPLinks = nil + wm.UserType = domain.UserTypeUnspecified wm.UserState = domain.UserStateDeleted } } diff --git a/internal/command/user_v2_invite_model.go b/internal/command/user_v2_invite_model.go index 91726498f..56545caa7 100644 --- a/internal/command/user_v2_invite_model.go +++ b/internal/command/user_v2_invite_model.go @@ -44,17 +44,31 @@ func (wm *UserV2InviteWriteModel) Reduce() error { for _, event := range wm.Events { switch e := event.(type) { case *user.HumanAddedEvent: - wm.UserState = domain.UserStateActive - wm.AuthMethodSet = crypto.SecretOrEncodedHash(e.Secret, e.EncodedHash) != "" wm.EmptyInviteCode() wm.ApplicationName = "" wm.AuthRequestID = "" + wm.URLTemplate = "" + wm.CodeReturned = false + wm.EmailVerified = false + wm.UserState = domain.UserStateActive + wm.AuthMethodSet = crypto.SecretOrEncodedHash(e.Secret, e.EncodedHash) != "" case *user.HumanRegisteredEvent: - wm.UserState = domain.UserStateActive - wm.AuthMethodSet = crypto.SecretOrEncodedHash(e.Secret, e.EncodedHash) != "" wm.EmptyInviteCode() wm.ApplicationName = "" wm.AuthRequestID = "" + wm.URLTemplate = "" + wm.CodeReturned = false + wm.EmailVerified = false + wm.UserState = domain.UserStateActive + wm.AuthMethodSet = crypto.SecretOrEncodedHash(e.Secret, e.EncodedHash) != "" + case *user.MachineAddedEvent: + wm.EmptyInviteCode() + wm.ApplicationName = "" + wm.AuthRequestID = "" + wm.URLTemplate = "" + wm.CodeReturned = false + wm.EmailVerified = false + wm.AuthMethodSet = false case *user.HumanInviteCodeAddedEvent: wm.SetInviteCode(e.Code, e.Expiry, e.CreationDate()) wm.URLTemplate = e.URLTemplate @@ -83,6 +97,13 @@ func (wm *UserV2InviteWriteModel) Reduce() error { case *user.UserReactivatedEvent: wm.UserState = domain.UserStateActive case *user.UserRemovedEvent: + wm.EmptyInviteCode() + wm.ApplicationName = "" + wm.AuthRequestID = "" + wm.URLTemplate = "" + wm.CodeReturned = false + wm.EmailVerified = false + wm.AuthMethodSet = false wm.UserState = domain.UserStateDeleted case *user.HumanPasswordChangedEvent: wm.AuthMethodSet = true @@ -118,6 +139,7 @@ func (wm *UserV2InviteWriteModel) Query() *eventstore.SearchQueryBuilder { user.HumanAddedType, user.UserV1RegisteredType, user.HumanRegisteredType, + user.MachineAddedEventType, user.HumanInviteCodeAddedType, user.HumanInviteCheckSucceededType, user.HumanInviteCheckFailedType, diff --git a/internal/command/user_v2_model.go b/internal/command/user_v2_model.go index 3ef070c6f..ea67a93ae 100644 --- a/internal/command/user_v2_model.go +++ b/internal/command/user_v2_model.go @@ -218,8 +218,10 @@ func (wm *UserV2WriteModel) Reduce() error { for _, event := range wm.Events { switch e := event.(type) { case *user.HumanAddedEvent: + wm.reset() wm.reduceHumanAddedEvent(e) case *user.HumanRegisteredEvent: + wm.reset() wm.reduceHumanRegisteredEvent(e) case *user.HumanInitialCodeAddedEvent: @@ -248,6 +250,7 @@ func (wm *UserV2WriteModel) Reduce() error { } case *user.MachineAddedEvent: + wm.reset() wm.UserName = e.UserName wm.Name = e.Name wm.Description = e.Description @@ -301,6 +304,7 @@ func (wm *UserV2WriteModel) Reduce() error { wm.UserState = domain.UserStateActive case *user.UserRemovedEvent: + wm.reset() wm.UserState = domain.UserStateDeleted case *user.HumanPasswordHashUpdatedEvent: @@ -628,3 +632,45 @@ func (wm *UserV2WriteModel) IDPLinkByID(idpID, externalUserID string) (idx int, } return -1, nil } + +func (wm *UserV2WriteModel) reset() { + wm.UserName = "" + wm.Name = "" + wm.Description = "" + wm.AccessTokenType = domain.OIDCTokenTypeBearer + wm.ClientSecret = nil + wm.FirstName = "" + wm.LastName = "" + wm.NickName = "" + wm.DisplayName = "" + wm.PreferredLanguage = language.Und + wm.Gender = domain.GenderUnspecified + wm.Avatar = "" + wm.InitCode = nil + wm.InitCodeCreationDate = time.Time{} + wm.InitCodeExpiry = 0 + wm.InitCheckFailedCount = 0 + wm.PasswordEncodedHash = "" + wm.PasswordChangeRequired = false + wm.PasswordCode = nil + wm.PasswordCodeCreationDate = time.Time{} + wm.PasswordCodeExpiry = 0 + wm.PasswordCheckFailedCount = 0 + wm.PasswordCodeGeneratorID = "" + wm.PasswordCodeVerificationID = "" + wm.Email = "" + wm.IsEmailVerified = false + wm.EmailCode = nil + wm.EmailCodeCreationDate = time.Time{} + wm.EmailCodeExpiry = 0 + wm.EmailCheckFailedCount = 0 + wm.Phone = "" + wm.IsPhoneVerified = false + wm.PhoneCode = nil + wm.PhoneCodeCreationDate = time.Time{} + wm.PhoneCodeExpiry = 0 + wm.PhoneCheckFailedCount = 0 + wm.IDPLinks = nil + wm.Metadata = nil + wm.CreationDate = time.Time{} +} diff --git a/internal/command/user_v2_model_test.go b/internal/command/user_v2_model_test.go index 602553cef..a485e63f2 100644 --- a/internal/command/user_v2_model_test.go +++ b/internal/command/user_v2_model_test.go @@ -409,14 +409,14 @@ func TestCommandSide_userExistsWriteModel(t *testing.T) { ProcessedSequence: 0, ResourceOwner: "org1", }, - UserName: "username", - FirstName: "firstname", - LastName: "lastname", - DisplayName: "firstname lastname", - PreferredLanguage: language.English, - PasswordEncodedHash: "$plain$x$password", - PasswordChangeRequired: true, - Email: "email@test.ch", + UserName: "", + FirstName: "", + LastName: "", + DisplayName: "", + PreferredLanguage: language.Und, + PasswordEncodedHash: "", + PasswordChangeRequired: false, + Email: "", IsEmailVerified: false, UserState: domain.UserStateDeleted, }, @@ -455,9 +455,9 @@ func TestCommandSide_userExistsWriteModel(t *testing.T) { ProcessedSequence: 0, ResourceOwner: "org1", }, - UserName: "username", - Name: "name", - Description: "description", + UserName: "", + Name: "", + Description: "", AccessTokenType: domain.OIDCTokenTypeBearer, UserState: domain.UserStateDeleted, }, diff --git a/internal/eventstore/event.go b/internal/eventstore/event.go index 651dae6ef..35392c2cb 100644 --- a/internal/eventstore/event.go +++ b/internal/eventstore/event.go @@ -40,6 +40,17 @@ type Command interface { Fields() []*FieldOperation } +type EnforceResourceOwnerCommand interface { + Command + // EnforceResourceOwner indicates that the resource owner of the event does not respect the previous events and should be enforced by the eventstore. + EnforceResourceOwner() +} + +func ShouldEnforceResourceOwner(cmd Command) bool { + _, ok := cmd.(EnforceResourceOwnerCommand) + return ok +} + // Event is a stored activity type Event interface { action diff --git a/internal/eventstore/eventstore_pusher_test.go b/internal/eventstore/eventstore_pusher_test.go index 318cf1a37..ee6650de4 100644 --- a/internal/eventstore/eventstore_pusher_test.go +++ b/internal/eventstore/eventstore_pusher_test.go @@ -607,6 +607,70 @@ func TestEventstore_Push_ResourceOwner(t *testing.T) { } } +func TestEventstore_Push_ResourceOwner_ReusedAggregateID(t *testing.T) { + aggregateType := eventstore.AggregateType(t.Name()) + aggregateID := "510" + + for pusherName, pusher := range pushers { + t.Run(pusherName, func(t *testing.T) { + t.Cleanup(cleanupEventstore(clients[pusherName])) + + db := eventstore.NewEventstore( + &eventstore.Config{ + Querier: queriers["v2(inmemory)"], + Pusher: pusher, + }, + ) + + initialEvents, err := db.Push(context.Background(), + generateCommand( + aggregateType, + aggregateID, + func(e *testEvent) { e.Agg.ResourceOwner = "old" }, + withEventType("test.initial"), + ), + ) + if err != nil { + t.Fatalf("initial eventstore.Push() error = %v", err) + } + if len(initialEvents) != 1 { + t.Fatalf("initial events length = %d, want 1", len(initialEvents)) + } + if initialEvents[0].Aggregate().ResourceOwner != "old" { + t.Fatalf("initial resource owner = %q, want %q", initialEvents[0].Aggregate().ResourceOwner, "old") + } + + reusedEvents, err := db.Push(context.Background(), + generateEnforcedCommand( + aggregateType, + aggregateID, + func(e *testEvent) { e.Agg.ResourceOwner = "new" }, + withEventType("test.recreated"), + ), + generateCommand( + aggregateType, + aggregateID, + func(e *testEvent) { e.Agg.ResourceOwner = "ignored" }, + withEventType("test.followup"), + ), + ) + if err != nil { + t.Fatalf("reused eventstore.Push() error = %v", err) + } + if len(reusedEvents) != 2 { + t.Fatalf("reused events length = %d, want 2", len(reusedEvents)) + } + for i, event := range reusedEvents { + if event.Aggregate().ResourceOwner != "new" { + t.Fatalf("reused event %d resource owner = %q, want %q", i, event.Aggregate().ResourceOwner, "new") + } + } + + assertResourceOwnersBySequence(t, clients[pusherName], []string{"old", "new", "new"}, string(aggregateType), aggregateID) + }) + } +} + func pushAggregates(es *eventstore.Eventstore, aggregateCommands [][]eventstore.Command) []error { wg := sync.WaitGroup{} errs := make([]error, 0) @@ -666,6 +730,33 @@ func assertResourceOwners(t *testing.T, db *database.DB, resourceOwners, aggrega } } +func assertResourceOwnersBySequence(t *testing.T, db *database.DB, resourceOwners []string, aggregateType, aggregateID string) { + t.Helper() + + eventCount := 0 + err := db.Query(func(rows *sql.Rows) error { + for i := 0; rows.Next(); i++ { + var resourceOwner string + if err := rows.Scan(&resourceOwner); err != nil { + return err + } + if resourceOwner != resourceOwners[i] { + t.Errorf("unexpected resource owner in queried event. want %q, got: %q", resourceOwners[i], resourceOwner) + } + eventCount++ + } + return nil + }, `SELECT owner FROM eventstore.events2 WHERE aggregate_type = $1 AND aggregate_id = $2 ORDER BY "sequence"`, aggregateType, aggregateID) + if err != nil { + t.Error("query failed: ", err) + return + } + + if eventCount != len(resourceOwners) { + t.Errorf("wrong queried event count: want %d, got %d", len(resourceOwners), eventCount) + } +} + func assertEventCount(t *testing.T, db *database.DB, aggTypes database.TextArray[eventstore.AggregateType], aggIDs database.TextArray[string], maxPushedEventsCount int) { t.Helper() diff --git a/internal/eventstore/local_postgres_test.go b/internal/eventstore/local_postgres_test.go index fdb8b4f51..5a9659bef 100644 --- a/internal/eventstore/local_postgres_test.go +++ b/internal/eventstore/local_postgres_test.go @@ -2,7 +2,6 @@ package eventstore_test import ( "context" - "encoding/json" "os" "testing" "time" @@ -159,11 +158,40 @@ func generateCommand(aggregateType eventstore.AggregateType, aggregateID string, return e } +func generateEnforcedCommand(aggregateType eventstore.AggregateType, aggregateID string, opts ...func(*testEvent)) eventstore.Command { + e := &enforcedTestEvent{ + testEvent: testEvent{ + BaseEvent: eventstore.BaseEvent{ + Agg: &eventstore.Aggregate{ + ID: aggregateID, + Type: aggregateType, + ResourceOwner: "ro", + Version: "v1", + }, + Service: "svc", + EventType: "test.created", + }, + }, + } + + for _, opt := range opts { + opt(&e.testEvent) + } + + return e +} + type testEvent struct { eventstore.BaseEvent uniqueConstraints []*eventstore.UniqueConstraint } +type enforcedTestEvent struct { + testEvent +} + +func (*enforcedTestEvent) EnforceResourceOwner() {} + func (e *testEvent) Payload() any { return e.BaseEvent.Data } @@ -207,13 +235,9 @@ func generateRemoveUniqueConstraint(table, uniqueField string) func(e *testEvent } } -func withTestData(data any) func(e *testEvent) { +func withEventType(eventType eventstore.EventType) func(e *testEvent) { return func(e *testEvent) { - d, err := json.Marshal(data) - if err != nil { - panic("marshal data failed") - } - e.BaseEvent.Data = d + e.EventType = eventType } } diff --git a/internal/eventstore/v3/event.go b/internal/eventstore/v3/event.go index c9ea4d2c6..38e2247bd 100644 --- a/internal/eventstore/v3/event.go +++ b/internal/eventstore/v3/event.go @@ -27,6 +27,7 @@ type command struct { Payload Payload Creator string Owner string + EnforceOwner bool } func (c *command) Aggregate() *eventstore.Aggregate { @@ -106,6 +107,7 @@ func commandToEvent(cmd eventstore.Command) (_ eventstore.Event, err error) { Payload: payload, Creator: cmd.Creator(), Owner: cmd.Aggregate().ResourceOwner, + EnforceOwner: eventstore.ShouldEnforceResourceOwner(cmd), } return &event{ diff --git a/internal/eventstore/v3/event_test.go b/internal/eventstore/v3/event_test.go index bd813c6ed..3f3398d92 100644 --- a/internal/eventstore/v3/event_test.go +++ b/internal/eventstore/v3/event_test.go @@ -96,6 +96,32 @@ func Test_commandToEvent(t *testing.T) { }, }, }, + { + name: "enforced resource owner", + args: args{ + command: &enforcedMockCommand{ + mockCommand: &mockCommand{ + aggregate: mockAggregate("V3-Red9I"), + payload: nil, + }, + }, + }, + want: want{ + event: &event{ + command: &command{ + InstanceID: "instance", + AggregateType: "type", + AggregateID: "V3-Red9I", + Owner: "ro", + Creator: "creator", + Revision: 1, + CommandType: "event.type", + Payload: nil, + EnforceOwner: true, + }, + }, + }, + }, } for _, tt := range tests { if tt.want.err == nil { @@ -454,6 +480,53 @@ func Test_commandsToEvents(t *testing.T) { }, }, }, + { + name: "single enforced command", + args: args{ + ctx: ctx, + cmds: []eventstore.Command{ + &enforcedMockCommand{ + mockCommand: &mockCommand{ + aggregate: mockAggregate("V3-Red9I"), + payload: nil, + }, + }, + }, + }, + want: want{ + events: []eventstore.Event{ + &event{ + command: &command{ + InstanceID: "instance", + AggregateType: "type", + AggregateID: "V3-Red9I", + Owner: "ro", + Creator: "creator", + Revision: 1, + CommandType: "event.type", + Payload: nil, + EnforceOwner: true, + }, + }, + }, + commands: []*command{ + { + InstanceID: "instance", + AggregateType: "type", + AggregateID: "V3-Red9I", + Owner: "ro", + CommandType: "event.type", + Revision: 1, + Payload: nil, + Creator: "creator", + EnforceOwner: true, + }, + }, + err: func(t *testing.T, err error) { + require.NoError(t, err) + }, + }, + }, } for _, tt := range tests { t.Run(tt.name, func(t *testing.T) { @@ -479,4 +552,5 @@ func assertCommand(t *testing.T, want, got *command) { assert.Equal(t, want.AggregateType, got.AggregateType) assert.Equal(t, want.InstanceID, got.InstanceID) assert.Equal(t, want.Revision, got.Revision) + assert.Equal(t, want.EnforceOwner, got.EnforceOwner) } diff --git a/internal/eventstore/v3/eventstore.go b/internal/eventstore/v3/eventstore.go index 884cb3e5a..82df93efc 100644 --- a/internal/eventstore/v3/eventstore.go +++ b/internal/eventstore/v3/eventstore.go @@ -98,11 +98,83 @@ var ( ElementType: commandType, }, } + command2Type = &pgtype.Type{ + Codec: &pgtype.CompositeCodec{ + Fields: []pgtype.CompositeCodecField{ + { + Name: "instance_id", + Type: textType, + }, + { + Name: "aggregate_type", + Type: textType, + }, + { + Name: "aggregate_id", + Type: textType, + }, + { + Name: "command_type", + Type: textType, + }, + { + Name: "revision", + Type: &pgtype.Type{ + Name: "int2", + OID: pgtype.Int2OID, + Codec: pgtype.Int2Codec{}, + }, + }, + { + Name: "payload", + Type: &pgtype.Type{ + Name: "jsonb", + OID: pgtype.JSONBOID, + Codec: &pgtype.JSONBCodec{ + Marshal: json.Marshal, + Unmarshal: json.Unmarshal, + }, + }, + }, + { + Name: "creator", + Type: textType, + }, + { + Name: "owner", + Type: textType, + }, + { + Name: "enforce_owner", + Type: &pgtype.Type{ + Name: "bool", + OID: pgtype.BoolOID, + Codec: pgtype.BoolCodec{}, + }, + }, + }, + }, + } + command2ArrayCodec = &pgtype.Type{ + Codec: &pgtype.ArrayCodec{ + ElementType: command2Type, + }, + } ) var typeMu sync.Mutex func RegisterEventstoreTypes(ctx context.Context, conn *pgx.Conn) error { + err := registerEventstoreType(ctx, conn, "command2", command2Type, command2ArrayCodec) + logging.OnError(err).Debug("failed to register command2") + + err = registerEventstoreType(ctx, conn, "command", commandType, commandArrayCodec) + logging.OnError(err).Debug("failed to register command") + + return nil +} + +func registerEventstoreType(ctx context.Context, conn *pgx.Conn, typ string, typeCodec, arrayCodec *pgtype.Type) error { // conn.TypeMap is not thread safe typeMu.Lock() defer typeMu.Unlock() @@ -114,14 +186,14 @@ func RegisterEventstoreTypes(ctx context.Context, conn *pgx.Conn) error { return nil } - if commandType.OID == 0 || commandArrayCodec.OID == 0 { - err := conn.QueryRow(ctx, "select oid, typarray from pg_type where typname = $1 and typnamespace = (select oid from pg_namespace where nspname = $2)", "command", "eventstore"). - Scan(&commandType.OID, &commandArrayCodec.OID) + if typeCodec.OID == 0 || arrayCodec.OID == 0 { + err := conn.QueryRow(ctx, "select oid, typarray from pg_type where typname = $1 and typnamespace = (select oid from pg_namespace where nspname = $2)", typ, "eventstore"). + Scan(&typeCodec.OID, &arrayCodec.OID) if err != nil { logging.WithError(err).Debug("failed to get oid for command type") return nil } - if commandType.OID == 0 || commandArrayCodec.OID == 0 { + if typeCodec.OID == 0 || arrayCodec.OID == 0 { logging.Debug("oid for command type not found") return nil } @@ -129,28 +201,28 @@ func RegisterEventstoreTypes(ctx context.Context, conn *pgx.Conn) error { m.RegisterTypes([]*pgtype.Type{ { - Name: "eventstore.command", - Codec: commandType.Codec, - OID: commandType.OID, + Name: "eventstore." + typ, + Codec: typeCodec.Codec, + OID: typeCodec.OID, }, { - Name: "command", - Codec: commandType.Codec, - OID: commandType.OID, + Name: typ, + Codec: typeCodec.Codec, + OID: typeCodec.OID, }, { - Name: "eventstore._command", - Codec: commandArrayCodec.Codec, - OID: commandArrayCodec.OID, + Name: "eventstore._" + typ, + Codec: arrayCodec.Codec, + OID: arrayCodec.OID, }, { - Name: "_command", - Codec: commandArrayCodec.Codec, - OID: commandArrayCodec.OID, + Name: "_" + typ, + Codec: arrayCodec.Codec, + OID: arrayCodec.OID, }, }) - dialect.RegisterDefaultPgTypeVariants[command](m, "eventstore.command", "eventstore._command") - dialect.RegisterDefaultPgTypeVariants[command](m, "command", "_command") + dialect.RegisterDefaultPgTypeVariants[command](m, "eventstore."+typ, "eventstore._"+typ) + dialect.RegisterDefaultPgTypeVariants[command](m, typ, "_"+typ) return nil } @@ -188,7 +260,7 @@ func CheckExecutionPlan(ctx context.Context, conn *sql.Conn) error { func (es *Eventstore) pushTx(ctx context.Context, client new_db.QueryExecutor) (tx new_db.Transaction, deferrable func(err error) error, err error) { tx, ok := client.(new_db.Transaction) if ok { - return tx, nil, nil + return tx, func(err error) error { return err }, nil } beginner, ok := client.(new_db.Beginner) if !ok { diff --git a/internal/eventstore/v3/mock_test.go b/internal/eventstore/v3/mock_test.go index fce586c4c..4a2e0dec8 100644 --- a/internal/eventstore/v3/mock_test.go +++ b/internal/eventstore/v3/mock_test.go @@ -5,6 +5,7 @@ import ( ) var _ eventstore.Command = (*mockCommand)(nil) +var _ eventstore.EnforceResourceOwnerCommand = (*enforcedMockCommand)(nil) type mockCommand struct { aggregate *eventstore.Aggregate @@ -12,6 +13,12 @@ type mockCommand struct { constraints []*eventstore.UniqueConstraint } +type enforcedMockCommand struct { + *mockCommand +} + +func (*enforcedMockCommand) EnforceResourceOwner() {} + // Aggregate implements [eventstore.Command] func (m *mockCommand) Aggregate() *eventstore.Aggregate { return m.aggregate diff --git a/internal/eventstore/v3/push.go b/internal/eventstore/v3/push.go index c31e9211a..ffae4246e 100644 --- a/internal/eventstore/v3/push.go +++ b/internal/eventstore/v3/push.go @@ -80,7 +80,12 @@ func writeEvents(ctx context.Context, tx database.Transaction, commands []events return nil, err } - rows, err := tx.Query(ctx, `select owner, created_at, "sequence", position from eventstore.push($1::eventstore.command[])`, cmds) + commandMapping := "eventstore.command2" + if command2Type.OID == 0 { + commandMapping = "eventstore.command" + } + + rows, err := tx.Query(ctx, fmt.Sprintf(`select owner, created_at, "sequence", position from eventstore.push($1::%s[])`, commandMapping), cmds) if err != nil { return nil, err } diff --git a/internal/eventstore/v3/sequence.go b/internal/eventstore/v3/sequence.go index 8e67b73e2..daac4e233 100644 --- a/internal/eventstore/v3/sequence.go +++ b/internal/eventstore/v3/sequence.go @@ -45,14 +45,7 @@ func latestSequences(ctx context.Context, tx database.Transaction, commands []ev } func searchSequenceByCommand(sequences []*latestSequence, command eventstore.Command) *latestSequence { - for _, sequence := range sequences { - if sequence.aggregate.Type == command.Aggregate().Type && - sequence.aggregate.ID == command.Aggregate().ID && - sequence.aggregate.InstanceID == command.Aggregate().InstanceID { - return sequence - } - } - return nil + return searchSequence(sequences, command.Aggregate().Type, command.Aggregate().ID, command.Aggregate().InstanceID) } func searchSequence(sequences []*latestSequence, aggregateType eventstore.AggregateType, aggregateID, instanceID string) *latestSequence { diff --git a/internal/eventstore/v3/sequence_test.go b/internal/eventstore/v3/sequence_test.go index d755c0dbd..8ec5794b0 100644 --- a/internal/eventstore/v3/sequence_test.go +++ b/internal/eventstore/v3/sequence_test.go @@ -217,6 +217,45 @@ func Test_commandsToSequences(t *testing.T) { }, }, }, + { + name: "first command enforcing keeps owner for aggregate", + args: args{ + ctx: context.Background(), + commands: []eventstore.Command{ + &enforcedMockCommand{ + mockCommand: &mockCommand{ + aggregate: &eventstore.Aggregate{ + ID: "V3-3oEV1", + Type: "type", + ResourceOwner: "new", + InstanceID: "instance", + Version: "v1", + }, + }, + }, + &mockCommand{ + aggregate: &eventstore.Aggregate{ + ID: "V3-3oEV1", + Type: "type", + ResourceOwner: "ignored", + InstanceID: "instance", + Version: "v1", + }, + }, + }, + }, + want: []*latestSequence{ + { + aggregate: &eventstore.Aggregate{ + ID: "V3-3oEV1", + Type: "type", + ResourceOwner: "new", + InstanceID: "instance", + Version: "v1", + }, + }, + }, + }, } for _, tt := range tests { t.Run(tt.name, func(t *testing.T) { diff --git a/internal/eventstore/write_model.go b/internal/eventstore/write_model.go index 965fb16d0..18d5f524c 100644 --- a/internal/eventstore/write_model.go +++ b/internal/eventstore/write_model.go @@ -38,6 +38,12 @@ func (wm *WriteModel) Reduce() error { wm.ResourceOwner = latestEvent.Aggregate().ResourceOwner } + for _, event := range wm.Events { + if _, ok := event.(EnforceResourceOwnerCommand); ok { + wm.ResourceOwner = event.Aggregate().ResourceOwner + } + } + if wm.InstanceID == "" { wm.InstanceID = latestEvent.Aggregate().InstanceID } diff --git a/internal/query/user_otp.go b/internal/query/user_otp.go index c25704813..5209f4195 100644 --- a/internal/query/user_otp.go +++ b/internal/query/user_otp.go @@ -72,6 +72,7 @@ func (wm *HumanOTPReadModel) Reduce() error { case *user.HumanOTPRemovedEvent: wm.State = domain.MFAStateRemoved case *user.UserRemovedEvent: + wm.Secret = nil wm.State = domain.MFAStateRemoved } } diff --git a/internal/query/user_password.go b/internal/query/user_password.go index 1d0037f72..4a60e5949 100644 --- a/internal/query/user_password.go +++ b/internal/query/user_password.go @@ -72,13 +72,29 @@ func (wm *HumanPasswordReadModel) Reduce() error { for _, event := range wm.Events { switch e := event.(type) { case *user.HumanAddedEvent: + wm.Code = nil + wm.CodeCreationDate = time.Time{} + wm.CodeExpiry = 0 + wm.PasswordCheckFailedCount = 0 wm.EncodedHash = crypto.SecretOrEncodedHash(e.Secret, e.EncodedHash) wm.SecretChangeRequired = e.ChangeRequired wm.UserState = domain.UserStateActive case *user.HumanRegisteredEvent: + wm.Code = nil + wm.CodeCreationDate = time.Time{} + wm.CodeExpiry = 0 + wm.PasswordCheckFailedCount = 0 wm.EncodedHash = crypto.SecretOrEncodedHash(e.Secret, e.EncodedHash) wm.SecretChangeRequired = e.ChangeRequired wm.UserState = domain.UserStateActive + case *user.MachineAddedEvent: + wm.Code = nil + wm.CodeCreationDate = time.Time{} + wm.CodeExpiry = 0 + wm.PasswordCheckFailedCount = 0 + wm.EncodedHash = "" + wm.SecretChangeRequired = false + wm.UserState = domain.UserStateUnspecified case *user.HumanInitialCodeAddedEvent: wm.UserState = domain.UserStateInitial case *user.HumanInitializedCheckSucceededEvent: @@ -108,6 +124,12 @@ func (wm *HumanPasswordReadModel) Reduce() error { wm.UserState = domain.UserStateActive } case *user.UserRemovedEvent: + wm.EncodedHash = "" + wm.SecretChangeRequired = false + wm.Code = nil + wm.CodeCreationDate = time.Time{} + wm.CodeExpiry = 0 + wm.PasswordCheckFailedCount = 0 wm.UserState = domain.UserStateDeleted case *user.HumanPasswordHashUpdatedEvent: wm.EncodedHash = e.EncodedHash @@ -124,6 +146,7 @@ func (wm *HumanPasswordReadModel) Query() *eventstore.SearchQueryBuilder { AggregateIDs(wm.AggregateID). EventTypes(user.HumanAddedType, user.HumanRegisteredType, + user.MachineAddedEventType, user.HumanInitialCodeAddedType, user.HumanInitializedCheckSucceededType, user.HumanPasswordChangedType, diff --git a/internal/repository/action/action.go b/internal/repository/action/action.go index 518238266..647d867a8 100644 --- a/internal/repository/action/action.go +++ b/internal/repository/action/action.go @@ -40,6 +40,8 @@ type AddedEvent struct { AllowedToFail bool `json:"allowedToFail"` } +func (*AddedEvent) EnforceResourceOwner() {} + func (e *AddedEvent) Payload() interface{} { return e } diff --git a/internal/repository/group/group.go b/internal/repository/group/group.go index 558cb8024..575bf97a4 100644 --- a/internal/repository/group/group.go +++ b/internal/repository/group/group.go @@ -20,6 +20,8 @@ type GroupAddedEvent struct { Description string `json:"description,omitempty"` } +func (*GroupAddedEvent) EnforceResourceOwner() {} + func NewGroupAddedEvent( ctx context.Context, aggregate *eventstore.Aggregate, diff --git a/internal/repository/project/project.go b/internal/repository/project/project.go index c86fed527..ba6fcd9c0 100644 --- a/internal/repository/project/project.go +++ b/internal/repository/project/project.go @@ -47,6 +47,8 @@ type ProjectAddedEvent struct { PrivateLabelingSetting domain.PrivateLabelingSetting `json:"privateLabelingSetting,omitempty"` } +func (*ProjectAddedEvent) EnforceResourceOwner() {} + func (e *ProjectAddedEvent) Payload() interface{} { return e } diff --git a/internal/repository/user/human.go b/internal/repository/user/human.go index 823a86aae..e0d6ad647 100644 --- a/internal/repository/user/human.go +++ b/internal/repository/user/human.go @@ -58,6 +58,8 @@ type HumanAddedEvent struct { ChangeRequired bool `json:"changeRequired,omitempty"` } +func (*HumanAddedEvent) EnforceResourceOwner() {} + func (e *HumanAddedEvent) Payload() interface{} { return e } @@ -167,6 +169,8 @@ type HumanRegisteredEvent struct { UserAgentID string `json:"userAgentID,omitempty"` } +func (*HumanRegisteredEvent) EnforceResourceOwner() {} + func (e *HumanRegisteredEvent) Payload() interface{} { return e } diff --git a/internal/repository/user/machine.go b/internal/repository/user/machine.go index b54149639..794941bf3 100644 --- a/internal/repository/user/machine.go +++ b/internal/repository/user/machine.go @@ -25,6 +25,8 @@ type MachineAddedEvent struct { AccessTokenType domain.OIDCTokenType `json:"accessTokenType,omitempty"` } +func (*MachineAddedEvent) EnforceResourceOwner() {} + func (e *MachineAddedEvent) Payload() interface{} { return e } diff --git a/internal/repository/user/schemauser/user.go b/internal/repository/user/schemauser/user.go index 6583e5736..651799f2a 100644 --- a/internal/repository/user/schemauser/user.go +++ b/internal/repository/user/schemauser/user.go @@ -26,6 +26,8 @@ type CreatedEvent struct { Data json.RawMessage `json:"user,omitempty"` } +func (*CreatedEvent) EnforceResourceOwner() {} + func (e *CreatedEvent) SetBaseEvent(event *eventstore.BaseEvent) { e.BaseEvent = event }