Packages
step_flow
1.6.1-rc0
1.9.0-rc2
1.9.0-rc1
1.9.0-rc0
1.8.2
1.8.1
1.8.1-rc8
1.8.1-rc7
1.8.1-rc6
1.8.1-rc5
1.8.1-rc4
1.8.1-rc3
1.8.1-rc2
1.8.1-rc1
1.8.1-rc0
1.8.0
1.8.0-rc3
1.8.0-rc2
1.8.0-rc1
1.8.0-rc0
1.7.3
1.7.3-rc4
1.7.3-rc3
1.7.3-rc2
1.7.3-rc1
1.7.3-rc0
1.7.2
1.7.2-rc4
1.7.2-rc3
1.7.2-rc2
1.7.2-rc1
1.7.2-rc0
1.7.1
1.7.0
1.7.0-rc1
1.7.0-rc0
1.6.1
1.6.1-rc1
1.6.1-rc0
1.6.0
1.6.0-rc9
1.6.0-rc8
1.6.0-rc7
1.6.0-rc6
1.6.0-rc5
1.6.0-rc4
1.6.0-rc3
1.6.0-rc20
1.6.0-rc2
1.6.0-rc19
1.6.0-rc18
1.6.0-rc17
1.6.0-rc16
1.6.0-rc15
1.6.0-rc14
1.6.0-rc13
1.6.0-rc12
1.6.0-rc11
1.6.0-rc10
1.6.0-rc1
1.5.0
1.5.0-rc1
1.4.2-rc2
1.4.2-rc1
1.4.1
1.4.1-rc1
1.4.0
1.4.0-rc4
1.4.0-rc3
1.4.0-rc2
1.4.0-rc1
1.3.1
1.3.0
1.3.0-rc
1.2.0
1.1.0
1.0.0
1.0.0-rc9
1.0.0-rc8
1.0.0-rc7
1.0.0-rc6
1.0.0-rc5
1.0.0-rc1
0.2.13
0.2.12
0.2.11
0.2.10
0.2.9
0.2.8
0.2.7
0.2.6
0.2.5
0.2.4
0.2.3
0.2.2
0.2.1
0.2.0
0.1.8
0.1.7
0.1.6
0.1.5
0.1.4
0.1.3
0.1.2
0.1.1
0.1.0
0.0.24
0.0.23
0.0.22
0.0.21
0.0.20
0.0.19
0.0.18
0.0.17
0.0.16
0.0.15
0.0.14
0.0.13
0.0.12
0.0.11
0.0.10
0.0.9
0.0.8
0.0.7
0.0.6
0.0.4
0.0.3
0.0.2
0.0.1
Step flow manager for Elixir applications
Current section
Files
Jump to
Current section
Files
lib/step_flow/notification_hooks/notification_hook_manager.ex
defmodule StepFlow.NotificationHooks.NotificationHookManager do
@moduledoc """
Notification manager context.
"""
alias StepFlow.Jobs
alias StepFlow.Workflows
require Logger
def get_last_status(workflow_notification_data) do
if StepFlow.Map.get_by_key_or_atom(workflow_notification_data, "status") == nil do
[]
else
StepFlow.Map.get_by_key_or_atom(workflow_notification_data, "status")
end
end
def notification_slack_from_job(job, channel, exposed_domain_name, description) do
send(
:step_flow_slack_bot,
{:message,
"Error for job #{job.name} ##{job.id} <#{exposed_domain_name}/workflows/#{job.workflow_id} |Open Workflow>\n```#{description}```",
channel}
)
end
def notification_teams_from_job(job, url, workflow, exposed_domain_name, description) do
headers = [{"content-type", "application/json"}]
body =
Poison.encode!(%{
"@context": "https://schema.org/extensions",
"@type": "MessageCard",
potentialAction: [
%{
"@type": "OpenUri",
name: "View Workflow",
targets: [
%{
os: "default",
uri: "#{exposed_domain_name}/workflows/#{job.workflow_id}"
}
]
}
],
sections: [
%{
facts: [
%{
name: "Posted By:",
value: "#{exposed_domain_name}"
},
%{
name: "Workflow Reference:",
value: "#{workflow.reference}"
},
%{
name: "Workflow ID:",
value: "#{workflow.id}"
},
%{
name: "Workflow Identifier:",
value: "#{workflow.identifier}"
},
%{
name: "Job ID:",
value: "#{job.id}"
},
%{
name: "Job:",
value: "#{job.name}"
}
],
text: "```#{description}```"
}
],
summary: "MCAI Workflow Error Notification",
themeColor: "FF5733",
title: "MCAI Workflow Error Notification"
})
Logger.debug(
"#{__MODULE__}: POST #{url}, headers: #{inspect(headers)}, body: #{inspect(body)}"
)
{:ok, response} = HTTPoison.request("POST", url, body, headers)
if response.status_code == 200 do
{:ok, response.body}
else
Logger.error("Unable to notify: #{inspect(response)}")
end
end
def notification_from_job(job_id, description \\ nil) do
job = Jobs.get_job!(job_id)
topic = "update_workflow_" <> Integer.to_string(job.workflow_id)
channel = StepFlow.Configuration.get_slack_channel()
workflow = Workflows.get_workflow!(job.workflow_id)
exposed_domain_name = StepFlow.Configuration.get_exposed_domain_name()
if StepFlow.Configuration.get_slack_token() != nil and description != nil and channel != nil do
notification_slack_from_job(job, channel, exposed_domain_name, description)
end
url = StepFlow.Configuration.get_teams_url()
if url != nil and description != nil do
notification_teams_from_job(job, url, workflow, exposed_domain_name, description)
end
StepFlow.Notification.send(topic, %{workflow_id: job.workflow_id})
end
def notify_endpoint_with_workflow_status(
workflow_notification_url,
workflow,
exposed_domain_name,
status_wf,
workflow_endpoint_credentials
) do
headers =
if workflow_endpoint_credentials != nil do
[
{"content-type", "application/json"},
{"Authorization", "Basic #{workflow_endpoint_credentials}"}
]
else
[{"content-type", "application/json"}]
end
body =
if status_wf == "workflow_error" do
Poison.encode!(%{
"@context": "https://schema.org/extensions",
"@type": "MessageCard",
potentialAction: [
%{
"@type": "OpenUri",
name: "View Workflow",
targets: [
%{
os: "default",
uri: "#{exposed_domain_name}/workflows/#{workflow.id}"
}
]
}
],
sections: [
%{
facts: [
%{
name: "Posted By",
value: "#{exposed_domain_name}"
},
%{
name: "Workflow Reference",
value: "#{workflow.reference}"
},
%{
name: "Workflow ID",
value: "#{workflow.id}"
},
%{
name: "Workflow Identifier",
value: "#{workflow.identifier}"
},
%{
name: "Workflow Status",
value: "#{status_wf}"
}
]
}
],
summary: "MCAI Workflow Error Notification",
themeColor: "FF5733",
title: "MCAI Workflow Error Notification"
})
else
Poison.encode!(%{
"@context": "https://schema.org/extensions",
"@type": "MessageCard",
potentialAction: [
%{
"@type": "OpenUri",
name: "View Workflow",
targets: [
%{
os: "default",
uri: "#{exposed_domain_name}/workflows/#{workflow.id}"
}
]
}
],
sections: [
%{
facts: [
%{
name: "Posted By",
value: "#{exposed_domain_name}"
},
%{
name: "Workflow Reference",
value: "#{workflow.reference}"
},
%{
name: "Workflow ID",
value: "#{workflow.id}"
},
%{
name: "Workflow Identifier",
value: "#{workflow.identifier}"
},
%{
name: "Workflow Status",
value: "#{status_wf}"
}
]
}
],
summary: "MCAI Workflow Success Notification",
themeColor: "339933",
title: "MCAI Workflow Success Notification"
})
end
Logger.debug(
"#{__MODULE__}: POST #{workflow_notification_url}, headers: #{inspect(headers)}, body: #{inspect(body)}"
)
{_, response} = HTTPoison.request("POST", workflow_notification_url, body, headers)
%{
"condition" => status_wf,
"timestamp" =>
NaiveDateTime.utc_now()
|> NaiveDateTime.to_string()
|> NaiveDateTime.from_iso8601!()
|> NaiveDateTime.truncate(:second),
"response" => response.status_code
}
end
def notify_endpoint_with_job_status(
workflow_notification_url,
job,
workflow,
exposed_domain_name,
status_job,
workflow_endpoint_credentials
) do
headers =
if workflow_endpoint_credentials != nil do
[
{"content-type", "application/json"},
{"Authorization", "Basic #{workflow_endpoint_credentials}"}
]
else
[{"content-type", "application/json"}]
end
body =
if status_job == "job_error" do
Poison.encode!(%{
"@context": "https://schema.org/extensions",
"@type": "MessageCard",
potentialAction: [
%{
"@type": "OpenUri",
name: "View Workflow",
targets: [
%{
os: "default",
uri: "#{exposed_domain_name}/workflows/#{workflow.id}"
}
]
}
],
sections: [
%{
facts: [
%{
name: "Posted By",
value: "#{exposed_domain_name}"
},
%{
name: "Workflow Reference",
value: "#{workflow.reference}"
},
%{
name: "Workflow ID",
value: "#{workflow.id}"
},
%{
name: "Job Id",
value: "#{job.id}"
},
%{
name: "Job Status",
value: "#{status_job}"
}
]
}
],
summary: "MCAI Job Error Notification",
themeColor: "FF5733",
title: "MCAI Job Error Notification"
})
else
Poison.encode!(%{
"@context": "https://schema.org/extensions",
"@type": "MessageCard",
potentialAction: [
%{
"@type": "OpenUri",
name: "View Workflow",
targets: [
%{
os: "default",
uri: "#{exposed_domain_name}/workflows/#{workflow.id}"
}
]
}
],
sections: [
%{
facts: [
%{
name: "Posted By",
value: "#{exposed_domain_name}"
},
%{
name: "Workflow Reference",
value: "#{workflow.reference}"
},
%{
name: "Workflow ID",
value: "#{workflow.id}"
},
%{
name: "Job Id",
value: "#{job.id}"
},
%{
name: "Job Status",
value: "#{status_job}"
}
]
}
],
summary: "MCAI Job Success Notification",
themeColor: "339933",
title: "MCAI Job Success Notification"
})
end
Logger.debug(
"#{__MODULE__}: POST #{workflow_notification_url}, headers: #{inspect(headers)}, body: #{inspect(body)}"
)
{_, response} = HTTPoison.request("POST", workflow_notification_url, body, headers)
%{
"condition" => status_job <> "_" <> Integer.to_string(job.id),
"timestamp" =>
NaiveDateTime.utc_now()
|> NaiveDateTime.to_string()
|> NaiveDateTime.from_iso8601!()
|> NaiveDateTime.truncate(:second),
"response" => response.status_code
}
end
def retrieve_status(
conditions,
job_status,
"job",
workflow_notification_data,
job,
workflow,
exposed_domain_name
) do
cond do
Enum.member?(
conditions,
"job_completed"
) and job_status == "completed" ->
notify_endpoint_with_job_status(
StepFlow.Map.get_by_key_or_atom(workflow_notification_data, "endpoint"),
job,
workflow,
exposed_domain_name,
"job_completed",
StepFlow.Map.get_by_key_or_atom(workflow_notification_data, "credentials")
)
Enum.member?(
conditions,
"job_error"
) and job_status == "error" ->
notify_endpoint_with_job_status(
StepFlow.Map.get_by_key_or_atom(workflow_notification_data, "endpoint"),
job,
workflow,
exposed_domain_name,
"job_error",
StepFlow.Map.get_by_key_or_atom(workflow_notification_data, "credentials")
)
true ->
nil
end
end
def retrieve_status(
conditions,
workflow_status,
"workflow",
workflow_notification_data,
workflow,
exposed_domain_name
) do
cond do
Enum.member?(
conditions,
"workflow_completed"
) and workflow_status == "completed" ->
notify_endpoint_with_workflow_status(
StepFlow.Map.get_by_key_or_atom(workflow_notification_data, "endpoint"),
workflow,
exposed_domain_name,
"workflow_completed",
StepFlow.Map.get_by_key_or_atom(workflow_notification_data, "credentials")
)
Enum.member?(
conditions,
"workflow_error"
) and workflow_status == "error" ->
notify_endpoint_with_workflow_status(
StepFlow.Map.get_by_key_or_atom(workflow_notification_data, "endpoint"),
workflow,
exposed_domain_name,
"workflow_error",
StepFlow.Map.get_by_key_or_atom(workflow_notification_data, "credentials")
)
true ->
nil
end
end
def manage_notification_status(job_id, "job", job_status) do
job = Jobs.get_job!(job_id)
workflow = Workflows.get_workflow!(job.workflow_id)
exposed_domain_name = StepFlow.Configuration.get_exposed_domain_name()
workflow_notifications_with_status =
for workflow_notification_data <- workflow.notification_hooks do
conditions = StepFlow.Map.get_by_key_or_atom(workflow_notification_data, "conditions")
last_status = get_last_status(workflow_notification_data)
new_status =
retrieve_status(
conditions,
job_status,
"job",
workflow_notification_data,
job,
workflow,
exposed_domain_name
)
if new_status != nil do
Map.put(workflow_notification_data, "status", last_status ++ [new_status])
else
workflow_notification_data
end
end
{_, workflow} =
Workflows.update_workflow(workflow, %{
notification_hooks: workflow_notifications_with_status
})
workflow
end
def manage_notification_status(workflow_id, "workflow", workflow_status) do
workflow = Workflows.get_workflow!(workflow_id)
exposed_domain_name = StepFlow.Configuration.get_exposed_domain_name()
workflow_notifications_with_status =
for workflow_notification_data <- workflow.notification_hooks do
conditions = StepFlow.Map.get_by_key_or_atom(workflow_notification_data, "conditions")
last_status = get_last_status(workflow_notification_data)
new_status =
retrieve_status(
conditions,
workflow_status,
"workflow",
workflow_notification_data,
workflow,
exposed_domain_name
)
if new_status != nil do
Map.put(workflow_notification_data, "status", last_status ++ [new_status])
else
workflow_notification_data
end
end
{_, workflow} =
Workflows.update_workflow(workflow, %{
notification_hooks: workflow_notifications_with_status
})
workflow
end
end