Proca.Stage.Processing (proca v3.3.1)

For these cases:

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

Processing works in following way:

  1. Process supporter, then action.
  2. Process supporter.
  3. 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

This method emits an effect on transition.

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

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.

Rank supporter when entering

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 in a Processing state

Link to this section Functions

Link to this function

change_status(p)

Link to this function

changed_action(processing)

Link to this function

clear_transient(p)

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.

Link to this function

exchange_for(org, atom)

Link to this function

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

Link to this function

needs_lookup?(arg1, stage)

Link to this function

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.

Link to this function

process_pipeline(p)

Link to this function

processing_org_id(processing)

Link to this function

rank_supporter(p)

Rank supporter when entering:

  1. supporter confirm stage
  2. action confirm stage
  3. action delivery stage
Link to this function

routing_for(map)

Link to this function

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

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

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