Proca.Server.Processing (proca v3.3.1)

For these cases:

We receive Actions with Supporter or with unbound ref.
Action with supporter may have new or resolved supporter (by ref).
Action with unbound ref will be bound later.

Processing works in following way:

Process supporter, then action.
Process supporter.
ignore. This is a case where we store action for counts (share, tweet without any contact, and it might never arrive). On the other hand, it would be nice to have this later in CRM right?

State diagram below shows transitions while processing. A stands for Action, S for supporter. States are enumerated in ProcessingStatus, and supporter and action track its status separately.

    [ A(NEW) / nil ]
        | linking
        v
    [ A(NEW) / S(NEW) ]            <-----.  linking new action to rejected supporter
        | emit to supporter confirm      | 
        v                                |  
    [ A(NEW) / S(CONFIRMING)] -> [ A(REJECTED) / S(REJECTED) ] --> stop (and remove the cookie?!)
        | confirm 
        v
  ,->[ A(NEW) / S(ACCEPTED)]
  |     | emit to action confirm
n |     v
e |  [ A(CONFIRMING) / S(ACCEPTED)] -> [ A(REJECTED) / S(ACCEPTED)] --> stop
w |     | confirm
  |     v
  '--[ A(ACCEPTED) / S(ACCEPTED)] -> [ A(DELIVERED) / S(ACCEPTED)] --> stop
                                 emit

This mechanism is supposed to be able to run many times with same result if action and supporter bits do not change.

At the moment we do not support custom action confirmation - fully - we do not send actions to this queue and there is not routes in API to confirm or reject an action. This is a missing piece albeit for now did not proove necessary.

Link to this section Summary

Functions

change_status(action, action_status, supporter_status)

child_spec(init_arg)

Returns a specification to start this module under a supervisor.

clear_transient(action)

emit(action, entity, arg3)

This method emits an effect on transition.

exchange_for(org, atom1, atom2)

maybe_rank_supporter(changeset, arg2, arg3)

Rank supporter when entering

process(action)

process_async(action)

routing_for(map)

start_link(opts)

sync()

A noop sync method that lets you make sure all previous async messages were processed (used in testing)

transition(action, arg2)

This function implements the state machine for Action. It returns a changeset to update action/supporter (state) and atoms telling where to route action.

Link to this section Functions

change_status(action, action_status, supporter_status)

child_spec(init_arg)

Returns a specification to start this module under a supervisor.

See Supervisor.

clear_transient(action)

emit(action, entity, arg3)

Specs

emit(
  action :: %Proca.Action{
    __meta__: term(),
    action_page: term(),
    action_page_id: term(),
    action_type: term(),
    campaign: term(),
    campaign_id: term(),
    donation: term(),
    fields: term(),
    id: term(),
    inserted_at: term(),
    messages: term(),
    processing_status: term(),
    ref: term(),
    source: term(),
    source_id: term(),
    supporter: term(),
    supporter_id: term(),
    testing: term(),
    updated_at: term(),
    with_consent: term()
  },
  :action | :supporter,
  :confirm | :deliver
) :: :ok | :error

This method emits an effect on transition.

We send whole action data, because a different system will consume it straight from rabbitmq.

exchange_for(org, atom1, atom2)

maybe_rank_supporter(changeset, arg2, arg3)

Rank supporter when entering:

supporter confirm stage
action delivery stage

process(action)

Specs

process(
  action :: %Proca.Action{
    __meta__: term(),
    action_page: term(),
    action_page_id: term(),
    action_type: term(),
    campaign: term(),
    campaign_id: term(),
    donation: term(),
    fields: term(),
    id: term(),
    inserted_at: term(),
    messages: term(),
    processing_status: term(),
    ref: term(),
    source: term(),
    source_id: term(),
    supporter: term(),
    supporter_id: term(),
    testing: term(),
    updated_at: term(),
    with_consent: term()
  }
) :: :ok

process_async(action)

routing_for(map)

start_link(opts)

sync()

A noop sync method that lets you make sure all previous async messages were processed (used in testing)

transition(action, arg2)

Specs

transition(
  %Proca.Action{
    __meta__: term(),
    action_page: term(),
    action_page_id: term(),
    action_type: term(),
    campaign: term(),
    campaign_id: term(),
    donation: term(),
    fields: term(),
    id: term(),
    inserted_at: term(),
    messages: term(),
    processing_status: term(),
    ref: term(),
    source: term(),
    source_id: term(),
    supporter: term(),
    supporter_id: term(),
    testing: term(),
    updated_at: term(),
    with_consent: term()
  },
  %Proca.ActionPage{
    __meta__: term(),
    campaign: term(),
    campaign_id: term(),
    config: term(),
    delivery: term(),
    extra_supporters: term(),
    id: term(),
    inserted_at: term(),
    live: term(),
    locale: term(),
    name: term(),
    org: term(),
    org_id: term(),
    supporter_confirm_template: term(),
    thank_you_template: term(),
    updated_at: term()
  }
) ::
  :ok
  | {Ecto.Changeset.t(%Proca.Action{
       __meta__: term(),
       action_page: term(),
       action_page_id: term(),
       action_type: term(),
       campaign: term(),
       campaign_id: term(),
       donation: term(),
       fields: term(),
       id: term(),
       inserted_at: term(),
       messages: term(),
       processing_status: term(),
       ref: term(),
       source: term(),
       source_id: term(),
       supporter: term(),
       supporter_id: term(),
       testing: term(),
       updated_at: term(),
       with_consent: term()
     }), :action | :supporter, :confirm | :deliver}
  | {Ecto.Changeset.t(%Proca.Action{
       __meta__: term(),
       action_page: term(),
       action_page_id: term(),
       action_type: term(),
       campaign: term(),
       campaign_id: term(),
       donation: term(),
       fields: term(),
       id: term(),
       inserted_at: term(),
       messages: term(),
       processing_status: term(),
       ref: term(),
       source: term(),
       source_id: term(),
       supporter: term(),
       supporter_id: term(),
       testing: term(),
       updated_at: term(),
       with_consent: term()
     }), nil, nil}

This function implements the state machine for Action. It returns a changeset to update action/supporter (state) and atoms telling where to route action.