Proca.Stage.Processing (proca v3.4.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(p)

changed_action(processing)

clear_transient(p)

Only clear transient personal info if we have reached delivery stage

emit(p, chan)

This method emits an effect on transition.

exchange_for(org, atom)

lookup_detail(p)

If we are emitting to queue, do the lookup and modify supporter and/or action

needs_lookup?(arg1, stage)

preload(actions)

Prepare the action associated data - arg can be single action or a list. In usual case, this action is already preloaded, so there is no database access here.

process_pipeline(p)

processing_org_id(processing)

rank_supporter(p)

Rank supporter when entering

routing_for(map)

store!(map)

transition(map, arg2)

This function implements the state machine for Action. It returns a new states for action and supporter, as well a queue stage where to emit the action. {action_state, supporter_state, queue_stage}

wrap(action)

Wrap Action in a Processing state

Link to this section Functions

change_status(p)

changed_action(processing)

clear_transient(p)

Only clear transient personal info if we have reached delivery stage

emit(p, chan)

Specs

emit(Proca.Stage.Processing, AMQP.Channel | nil) :: :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, atom)

lookup_detail(p)

Specs

lookup_detail(%Proca.Stage.Processing{
  action_change: term(),
  details: term(),
  new_state: term(),
  stage: term(),
  supporter_change: term()
}) ::
  {:ok,
   %Proca.Stage.Processing{
     action_change: term(),
     details: term(),
     new_state: term(),
     stage: term(),
     supporter_change: term()
   }}
  | {:error, term()}

If we are emitting to queue, do the lookup and modify supporter and/or action

needs_lookup?(arg1, stage)

preload(actions)

Prepare the action associated data - arg can be single action or a list. In usual case, this action is already preloaded, so there is no database access here.

process_pipeline(p)

processing_org_id(processing)

rank_supporter(p)

Rank supporter when entering:

supporter confirm stage
action confirm stage
action delivery stage

routing_for(map)

store!(map)

transition(map, 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
  | {:new | :confirming | :accepted | :delivered,
     :new | :confirming | :accepted,
     :supporter_confirm | :action_confirm | :deliver | nil}

This function implements the state machine for Action. It returns a new states for action and supporter, as well a queue stage where to emit the action. {action_state, supporter_state, queue_stage}

returns :ok if nothing needs to be done

wrap(action)

Specs

wrap(%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()
}) ::
  :noop
  | {:lookup_detail,
     %Proca.Stage.Processing{
       action_change: term(),
       details: term(),
       new_state: term(),
       stage: term(),
       supporter_change: term()
     }}
  | {:process,
     %Proca.Stage.Processing{
       action_change: term(),
       details: term(),
       new_state: term(),
       stage: term(),
       supporter_change: term()
     }}

Wrap Action in a Processing state

Check which stage we are at
Check if we need detail lookup.
If so, run the lookup task and exit -> a continuation will come via handle_info() and we continue in process_pipeline
Else run straight process_pipeline