XYZ-447: Add routing affinity and email to Customer

apps/cs/priv/migrations/0000000000002-customer-email.sql

@@ -0,0 +1,7 @@
+ALTER TABLE customer ADD COLUMN IF NOT EXISTS email TEXT;
+
+-- email is stored already normalized (lower + trim), hence a plain column index
+-- rather than an expression one; partial, same as external_id
+CREATE UNIQUE INDEX IF NOT EXISTS idx_customer_email_party
+    ON customer(email, party_ref)
+    WHERE deleted_at IS NULL AND email IS NOT NULL;

apps/cs/priv/migrations/0000000000003-terminal-affinity.sql

@@ -0,0 +1,26 @@
+CREATE TABLE IF NOT EXISTS terminal_affinity (
+    id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
+    customer_id UUID NOT NULL REFERENCES customer(id),
+    provider_ref TEXT NOT NULL,
+    terminal_ref TEXT NOT NULL,
+    -- strict total order of bindings, immune to timestamp collisions
+    bind_seq BIGSERIAL NOT NULL,
+    bound_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
+    -- base for since_last_use; set on bind, bumped by every successful payment
+    last_used_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
+    released_at TIMESTAMPTZ,
+    released_reason TEXT
+);
+
+CREATE UNIQUE INDEX IF NOT EXISTS idx_terminal_affinity_unique
+    ON terminal_affinity(customer_id, provider_ref, terminal_ref)
+    WHERE released_at IS NULL;
+
+CREATE INDEX IF NOT EXISTS idx_terminal_affinity_lookup
+    ON terminal_affinity(customer_id, bind_seq)
+    WHERE released_at IS NULL;
+
+-- for the admin "release everyone off terminal X" operation
+CREATE INDEX IF NOT EXISTS idx_terminal_affinity_terminal
+    ON terminal_affinity(provider_ref, terminal_ref)
+    WHERE released_at IS NULL;

apps/cs/src/cs_customer.erl

@@ -1,11 +1,14 @@
 -module(cs_customer).
 
 -export([
-    create/4,
+    create/5,
     get/1,
     get_state/1,
     get_by_external_id/2,
+    get_by_email/2,
     get_by_payment/2,
+    find_or_create_by_email/2,
+    normalize_email/1,
     delete/1,
     add_bank_card/2,
     remove_bank_card/2,
@@ -15,14 +18,20 @@
 ]).
 
 -export_type([customer_id/0, bank_card_id/0, party_ref/0, contact_info/0, metadata/0]).
--export_type([customer/0, customer_state/0, payment_ref/0]).
+-export_type([customer/0, customer_state/0, payment_ref/0, email/0]).
+
+%% Reason recorded on affinities released together with the customer
+-define(DELETED_REASON, <<"customer_deleted">>).
+%% Longest address an SMTP envelope may carry (RFC 5321)
+-define(MAX_EMAIL_SIZE, 320).
 
 %% Types
 -type customer_id() :: binary().
 -type bank_card_id() :: binary().
 -type party_ref() :: dmsl_domain_thrift:'PartyConfigRef'().
 -type contact_info() :: dmsl_domain_thrift:'ContactInfo'() | undefined.
 -type metadata() :: dmsl_domain_thrift:'Metadata'() | undefined.
+-type email() :: binary().
 
 -type customer() :: #{
     id := customer_id(),
@@ -31,7 +40,8 @@
     metadata => metadata(),
     created_at := binary(),
     deleted_at => binary() | undefined,
-    external_id => binary() | undefined
+    external_id => binary() | undefined,
+    email => email() | undefined
 }.
 -type customer_state() :: #{
     customer := customer(),
@@ -46,11 +56,18 @@
 
 %% API
 
--spec create(party_ref(), contact_info(), metadata(), binary() | undefined) ->
-    {ok, customer_id()} | {error, term()}.
-create(PartyRef, ContactInfo, Metadata, ExternalID) ->
-    PartyRefJson = party_ref_to_json(PartyRef),
-    cs_customer_database:create(PartyRefJson, ContactInfo, Metadata, ExternalID).
+-spec create(party_ref(), contact_info(), metadata(), binary() | undefined, email() | undefined) ->
+    {ok, customer_id()} | {error, external_id_conflict | email_conflict | invalid_email | term()}.
+create(PartyRef, ContactInfo, Metadata, ExternalID, Email) ->
+    case maybe_normalize_email(Email) of
+        {ok, NormalizedEmail} ->
+            PartyRefJson = party_ref_to_json(PartyRef),
+            cs_customer_database:create(
+                PartyRefJson, ContactInfo, Metadata, ExternalID, NormalizedEmail
+            );
+        {error, _} = Error ->
+            Error
+    end.
 
 -spec get(customer_id()) -> {ok, customer()} | {error, not_found | term()}.
 get(CustomerID) ->
@@ -95,6 +112,33 @@ get_by_external_id(ExternalID, PartyRef) ->
             Error
     end.
 
+-spec get_by_email(party_ref(), email()) ->
+    {ok, customer_state()} | {error, not_found | invalid_email | term()}.
+get_by_email(PartyRef, Email) ->
+    case normalize_email(Email) of
+        {ok, NormalizedEmail} ->
+            PartyRefJson = party_ref_to_json(PartyRef),
+            case cs_customer_database:get_by_email(NormalizedEmail, PartyRefJson) of
+                {ok, Customer} ->
+                    get_state(maps:get(id, Customer));
+                Error ->
+                    Error
+            end;
+        {error, _} = Error ->
+            Error
+    end.
+
+-spec find_or_create_by_email(party_ref(), email()) ->
+    {ok, customer()} | {error, invalid_email | term()}.
+find_or_create_by_email(PartyRef, Email) ->
+    case normalize_email(Email) of
+        {ok, NormalizedEmail} ->
+            PartyRefJson = party_ref_to_json(PartyRef),
+            cs_customer_database:find_or_create_by_email(PartyRefJson, NormalizedEmail);
+        {error, _} = Error ->
+            Error
+    end.
+
 -spec get_by_payment(binary(), binary()) ->
     {ok, customer_state()} | {error, not_found | invalid_recurrent_parent | term()}.
 get_by_payment(InvoiceId, PaymentId) ->
@@ -108,9 +152,28 @@ get_by_payment(InvoiceId, PaymentId) ->
             Error
     end.
 
+-spec normalize_email(email()) -> {ok, email()} | {error, invalid_email}.
+normalize_email(Email) when is_binary(Email) ->
+    try unicode:characters_to_binary(string:trim(string:lowercase(Email))) of
+        Normalized when is_binary(Normalized) -> validate_email(Normalized);
+        _ -> {error, invalid_email}
+    catch
+        %% string:lowercase/1 fails with badarg on malformed UTF-8
+        _:_ -> {error, invalid_email}
+    end;
+normalize_email(_Email) ->
+    {error, invalid_email}.
+
 -spec delete(customer_id()) -> ok | {error, not_found | term()}.
 delete(CustomerID) ->
-    cs_customer_database:delete(CustomerID).
+    case cs_customer_database:delete(CustomerID) of
+        ok ->
+            %% Orphan affinities would outlive the customer and the freed email
+            _ = release_terminal_affinities(CustomerID),
+            ok;
+        Error ->
+            Error
+    end.
 
 -spec add_bank_card(customer_id(), bank_card_id()) -> ok | {error, customer_not_found | term()}.
 add_bank_card(CustomerID, BankCardId) ->
@@ -157,6 +220,45 @@ get_payments(CustomerID, Limit, Offset) ->
 
 %% Internal functions
 
+-spec maybe_normalize_email(email() | undefined) -> {ok, email() | undefined} | {error, invalid_email}.
+maybe_normalize_email(undefined) -> {ok, undefined};
+maybe_normalize_email(Email) -> normalize_email(Email).
+
+-spec validate_email(binary()) -> {ok, email()} | {error, invalid_email}.
+validate_email(<<>>) ->
+    {error, invalid_email};
+%% Beyond the RFC 5321 envelope limit the value is not an address, and it also stops
+%% fitting into idx_customer_email_party (btree rejects entries over ~2704 bytes)
+validate_email(Email) when byte_size(Email) > ?MAX_EMAIL_SIZE ->
+    {error, invalid_email};
+validate_email(Email) ->
+    case binary:match(Email, <<"@">>) of
+        nomatch -> {error, invalid_email};
+        _ -> validate_email_charset(Email)
+    end.
+
+%% Control characters never belong in an address, and a NUL byte cannot be stored in a
+%% text column at all, so PostgreSQL would answer with an error instead of a conflict
+-spec validate_email_charset(binary()) -> {ok, email()} | {error, invalid_email}.
+validate_email_charset(Email) ->
+    ControlChars = [<<Char>> || Char <- lists:seq(0, 31)] ++ [<<127>>],
+    case binary:match(Email, ControlChars) of
+        nomatch -> {ok, Email};
+        _ -> {error, invalid_email}
+    end.
+
+-spec release_terminal_affinities(customer_id()) -> ok.
+release_terminal_affinities(CustomerID) ->
+    case cs_terminal_affinity:release_by_customer(CustomerID, ?DELETED_REASON) of
+        ok ->
+            ok;
+        {error, Reason} ->
+            logger:warning("failed to release terminal affinities of customer ~s: ~p", [
+                CustomerID, Reason
+            ]),
+            ok
+    end.
+
 -spec make_continuation_token(non_neg_integer(), non_neg_integer(), non_neg_integer()) ->
     binary() | undefined.
 make_continuation_token(Offset, Count, Total) ->

apps/cs/src/cs_customer_database.erl

@@ -1,9 +1,11 @@
 -module(cs_customer_database).
 
 -export([
-    create/4,
+    create/5,
     get/1,
     get_by_external_id/2,
+    get_by_email/2,
+    find_or_create_by_email/2,
     get_by_payment/2,
     delete/1,
     link_bank_card/2,
@@ -28,27 +30,29 @@
 
 %% API
 
--spec create(binary(), term(), term(), binary() | undefined) ->
-    {ok, customer_id()} | {error, term()}.
-create(PartyRef, ContactInfo, Metadata, ExternalID) ->
+-spec create(binary(), term(), term(), binary() | undefined, binary() | undefined) ->
+    {ok, customer_id()} | {error, external_id_conflict | email_conflict | term()}.
+create(PartyRef, ContactInfo, Metadata, ExternalID, Email) ->
     Query = """
-    INSERT INTO customer (party_ref, contact_info, metadata, external_id)
-    VALUES ($1, $2, $3, $4)
+    INSERT INTO customer (party_ref, contact_info, metadata, external_id, email)
+    VALUES ($1, $2, $3, $4, $5)
     RETURNING id
     """,
-    Params = [PartyRef, encode_contact_info(ContactInfo), encode_metadata(Metadata), ExternalID],
+    Params = [
+        PartyRef, encode_contact_info(ContactInfo), encode_metadata(Metadata), ExternalID, Email
+    ],
     case epg_pool:query(?POOL, Query, Params) of
         {ok, _, _, [{Id}]} -> {ok, Id};
         {ok, _, [{Id}]} -> {ok, Id};
-        {error, #error{codename = unique_violation}} -> {error, external_id_conflict};
+        {error, #error{codename = unique_violation, extra = Extra}} -> unique_violation(Extra);
         {error, Reason} -> {error, Reason};
         Other -> {error, {unexpected_result, Other}}
     end.
 
 -spec get(customer_id()) -> {ok, customer()} | {error, not_found | term()}.
 get(CustomerID) ->
     Query = """
-    SELECT id, party_ref, contact_info, metadata, created_at, deleted_at, external_id
+    SELECT id, party_ref, contact_info, metadata, created_at, deleted_at, external_id, email
     FROM customer
     WHERE id = $1::uuid
     """,
@@ -63,7 +67,7 @@ get(CustomerID) ->
 -spec get_by_external_id(binary(), binary()) -> {ok, customer()} | {error, not_found | term()}.
 get_by_external_id(ExternalID, PartyRef) ->
     Query = """
-    SELECT id, party_ref, contact_info, metadata, created_at, deleted_at, external_id
+    SELECT id, party_ref, contact_info, metadata, created_at, deleted_at, external_id, email
     FROM customer
     WHERE external_id = $1
       AND party_ref = $2
@@ -78,10 +82,48 @@ get_by_external_id(ExternalID, PartyRef) ->
         {error, Reason} -> {error, Reason}
     end.
 
+%% Email is expected to be already normalized by the business layer.
+-spec get_by_email(binary(), binary()) -> {ok, customer()} | {error, not_found | term()}.
+get_by_email(Email, PartyRef) ->
+    Query = """
+    SELECT id, party_ref, contact_info, metadata, created_at, deleted_at, external_id, email
+    FROM customer
+    WHERE email = $1
+      AND party_ref = $2
+      AND deleted_at IS NULL
+    LIMIT 1
+    """,
+    case epg_pool:query(?POOL, Query, [Email, PartyRef]) of
+        {ok, _, _, [Row]} -> {ok, row_to_customer(Row)};
+        {ok, _, _, []} -> {error, not_found};
+        {ok, _, [Row]} -> {ok, row_to_customer(Row)};
+        {ok, _, []} -> {error, not_found};
+        {error, Reason} -> {error, Reason}
+    end.
+
+-spec find_or_create_by_email(binary(), binary()) -> {ok, customer()} | {error, term()}.
+find_or_create_by_email(PartyRef, Email) ->
+    %% ON CONFLICT predicate is literally the predicate of idx_customer_email_party,
+    %% otherwise PostgreSQL fails to infer the arbiter index (42P10). The DO UPDATE is
+    %% a no-op needed to make RETURNING fire for an already existing row as well.
+    Query = """
+    INSERT INTO customer (party_ref, email)
+    VALUES ($1, $2)
+    ON CONFLICT (email, party_ref) WHERE deleted_at IS NULL AND email IS NOT NULL
+    DO UPDATE SET email = EXCLUDED.email
+    RETURNING id, party_ref, contact_info, metadata, created_at, deleted_at, external_id, email
+    """,
+    case query_rows(?POOL, Query, [PartyRef, Email]) of
+        {ok, [Row]} -> {ok, row_to_customer(Row)};
+        {ok, []} -> {error, failed_to_create};
+        {error, Reason} -> {error, Reason}
+    end.
+
 -spec get_by_payment(invoice_id(), payment_id()) -> {ok, customer()} | {error, not_found | term()}.
 get_by_payment(InvoiceId, PaymentId) ->
     Query = """
-    SELECT c.id, c.party_ref, c.contact_info, c.metadata, c.created_at, c.deleted_at, c.external_id
+    SELECT c.id, c.party_ref, c.contact_info, c.metadata, c.created_at, c.deleted_at,
+           c.external_id, c.email
     FROM customer c
     JOIN payment_ref pr ON c.id = pr.customer_id
     WHERE pr.invoice_id = $1
@@ -191,17 +233,27 @@ get_payments(CustomerID, Limit, Offset) ->
 
 %% Internal functions
 
-row_to_customer({Id, PartyRef, ContactInfo, Metadata, CreatedAt, DeletedAt, ExternalID}) ->
+row_to_customer({Id, PartyRef, ContactInfo, Metadata, CreatedAt, DeletedAt, ExternalID, Email}) ->
     #{
         id => Id,
         party_ref => PartyRef,
         contact_info => decode_contact_info(ContactInfo),
         metadata => decode_metadata(Metadata),
         created_at => CreatedAt,
         deleted_at => null_to_default(DeletedAt, undefined),
-        external_id => null_to_default(ExternalID, undefined)
+        external_id => null_to_default(ExternalID, undefined),
+        email => null_to_default(Email, undefined)
     }.
 
+%% Both partial unique indexes on customer are reachable from create/5, so the
+%% conflicting one is told apart by constraint name.
+unique_violation(Extra) ->
+    case proplists:get_value(constraint_name, Extra) of
+        <<"idx_customer_external_id_party">> -> {error, external_id_conflict};
+        <<"idx_customer_email_party">> -> {error, email_conflict};
+        Name -> {error, {unique_violation, Name}}
+    end.
+
 null_to_default(null, Default) -> Default;
 null_to_default(V, _Default) -> V.