init commit

lib/workload_service/aggregates/quote_task.ex

@@ -0,0 +1,9 @@
+defmodule WorkloadService.Aggregates.QuoteTask do
+  use WorkloadService.Aggregates.Task,
+    task_type: "quote",
+    commands: WorkloadService.Commands.QuoteTask
+
+  def validate_submission(_) do
+    :ok
+  end
+end

lib/workload_service/aggregates/solicitation_task.ex

@@ -0,0 +1,9 @@
+defmodule WorkloadService.Aggregates.SolicitationTask do
+  use WorkloadService.Aggregates.Task,
+    task_type: "solicitation",
+    commands: WorkloadService.Commands.SolicitationTask
+
+  def validate_submission(_) do
+    :ok
+  end
+end

lib/workload_service/aggregates/task.ex

@@ -0,0 +1,145 @@
+defmodule WorkloadService.Aggregates.Task do
+  @moduledoc """
+  Behaviour and __using__ macro for task aggregates.
+  Each task type can override validation and add specific behavior.
+
+  Usage:
+    defmodule WorkloadService.Aggregates.QuoteTask do
+      use WorkloadService.Aggregates.Task,
+        task_type: "quote"
+    end
+  """
+
+  @callback validate_submission(map()) :: :ok | {:error, term()}
+
+  defmacro __using__(opts) do
+    task_type = Keyword.fetch!(opts, :task_type)
+    commands_module = Keyword.get(opts, :commands, WorkloadService.Commands.Task)
+
+    quote do
+      @behaviour Commanded.Aggregates.Aggregate
+      @task_type unquote(task_type)
+
+      alias unquote(commands_module).CreateTask
+      alias unquote(commands_module).SubmitResponse
+      alias unquote(commands_module).ApproveSubmission
+      alias unquote(commands_module).CompleteTask
+
+      alias Commanded.Aggregates.Aggregate
+
+      defstruct [
+        :id,
+        :application_id,
+        :provider_id,
+        :provider_name,
+        :task_info,
+        :submission,
+        :attachments,
+        :status,
+        :version
+      ]
+
+      @impl Aggregate
+      def execute(%__MODULE__{status: nil}, %CreateTask{} = cmd) do
+        %WorkloadService.Events.TaskCreated{
+          id: cmd.id,
+          application_id: cmd.application_id,
+          provider_id: cmd.provider_id,
+          provider_name: cmd.provider_name,
+          task_info: cmd.task_info || %{},
+          attachments: cmd.attachments || []
+        }
+      end
+
+      @impl Aggregate
+      def execute(%__MODULE__{}, %CreateTask{}) do
+        {:error, :task_already_exists}
+      end
+
+      @impl Aggregate
+      def execute(%__MODULE__{status: status}, %SubmitResponse{} = cmd)
+          when status in [nil, "created", "draft", "approved"] do
+        with :ok <- validate_submission(cmd.submission) do
+          new_status = if status == "approved", do: "draft", else: "draft"
+
+          %WorkloadService.Events.SubmissionUpdated{
+            id: cmd.id,
+            submission: cmd.submission,
+            attachments: cmd.attachments
+          }
+        end
+      end
+
+      @impl Aggregate
+      def execute(%__MODULE__{status: "draft"}, %ApproveSubmission{}) do
+        %WorkloadService.Events.SubmissionApproved{
+          id: nil
+        }
+      end
+
+      @impl Aggregate
+      def execute(%__MODULE__{status: status}, %ApproveSubmission{}) do
+        {:error, {:invalid_state, "cannot approve in state: #{status}"}}
+      end
+
+      @impl Aggregate
+      def execute(%__MODULE__{status: "approved"}, %CompleteTask{} = cmd) do
+        %WorkloadService.Events.TaskCompleted{
+          id: cmd.id,
+          completed_by: cmd.completed_by
+        }
+      end
+
+      @impl Aggregate
+      def execute(%__MODULE__{status: status}, %CompleteTask{}) do
+        {:error, {:invalid_state, "cannot complete in state: #{status}"}}
+      end
+
+      @impl Aggregate
+      def apply(%__MODULE__{} = agg, %WorkloadService.Events.TaskCreated{} = e) do
+        %{
+          agg
+          | id: e.id,
+            application_id: e.application_id,
+            provider_id: e.provider_id,
+            provider_name: e.provider_name,
+            task_info: e.task_info,
+            attachments: e.attachments,
+            status: "created",
+            version: agg.version + 1
+        }
+      end
+
+      @impl Aggregate
+      def apply(%__MODULE__{} = agg, %WorkloadService.Events.SubmissionUpdated{} = e) do
+        %{
+          agg
+          | submission: e.submission,
+            attachments: e.attachments || [],
+            status: "draft",
+            version: agg.version + 1
+        }
+      end
+
+      @impl Aggregate
+      def apply(%__MODULE__{} = agg, %WorkloadService.Events.SubmissionApproved{}) do
+        %{
+          agg
+          | status: "approved",
+            version: agg.version + 1
+        }
+      end
+
+      @impl Aggregate
+      def apply(%__MODULE__{} = agg, %WorkloadService.Events.TaskCompleted{}) do
+        %{
+          agg
+          | status: "completed",
+            version: agg.version + 1
+        }
+      end
+
+      defoverridable execute: 2, apply: 2
+    end
+  end
+end

lib/workload_service/aggregates/task_id.ex

@@ -0,0 +1,48 @@
+defmodule WorkloadService.Aggregates.TaskId do
+  @moduledoc """
+  Task identifier with org_id, type and task_id.
+  ID format: "org_id:type:task_id" (e.g., "test:quote:uuid")
+  """
+
+  @derive Jason.Encoder
+  defstruct [:org_id, :type, :task_id]
+
+  def new(org_id, type, task_id) when type in ["quote", "solicitation"] do
+    %__MODULE__{
+      org_id: org_id,
+      type: type,
+      task_id: task_id
+    }
+  end
+
+  def parse(string) when is_binary(string) do
+    case String.split(string, ":", parts: 3) do
+      [org_id, type, task_id] when type in ["quote", "solicitation"] ->
+        {:ok, new(org_id, type, task_id)}
+
+      _ ->
+        {:error, :invalid_task_id}
+    end
+  end
+
+  def parse!(string) do
+    case parse(string) do
+      {:ok, id} -> id
+      {:error, reason} -> raise ArgumentError, "invalid task id #{inspect(string)}: #{reason}"
+    end
+  end
+
+  defimpl String.Chars do
+    def to_string(%WorkloadService.Aggregates.TaskId{
+          org_id: org_id,
+          type: type,
+          task_id: task_id
+        }) do
+      "#{org_id}:#{type}:#{task_id}"
+    end
+  end
+
+  defimpl Commanded.Serialization.JsonDecoder do
+    def decode(id), do: id
+  end
+end

lib/workload_service/application.ex

@@ -0,0 +1,29 @@
+defmodule WorkloadService.Application do
+  @moduledoc false
+
+  use Application
+
+  @impl true
+  def start(_type, _args) do
+    children = [
+      WorkloadService.CommandedApp,
+      WorkloadService.Consumers.QuoteRequestedConsumer,
+      WorkloadService.Consumers.SolicitationRequestedConsumer,
+      WorkloadService.Projectors.TaskProjector,
+      WorkloadService.Repo,
+      WorkloadServiceWeb.Telemetry,
+      {DNSCluster, query: Application.get_env(:workload_service, :dns_cluster_query) || :ignore},
+      {Phoenix.PubSub, name: WorkloadService.PubSub},
+      WorkloadServiceWeb.Endpoint
+    ]
+
+    opts = [strategy: :one_for_one, name: WorkloadService.Supervisor]
+    Supervisor.start_link(children, opts)
+  end
+
+  @impl true
+  def config_change(changed, _new, removed) do
+    WorkloadServiceWeb.Endpoint.config_change(changed, removed)
+    :ok
+  end
+end

lib/workload_service/commanded_app.ex

@@ -0,0 +1,32 @@
+defmodule WorkloadService.Router do
+  use Commanded.Commands.Router
+
+  dispatch(
+    [
+      WorkloadService.Commands.QuoteTask.CreateTask,
+      WorkloadService.Commands.QuoteTask.SubmitResponse,
+      WorkloadService.Commands.QuoteTask.ApproveSubmission,
+      WorkloadService.Commands.QuoteTask.CompleteTask
+    ],
+    to: WorkloadService.Aggregates.QuoteTask,
+    identity: :id
+  )
+
+  dispatch(
+    [
+      WorkloadService.Commands.SolicitationTask.CreateTask,
+      WorkloadService.Commands.SolicitationTask.SubmitResponse,
+      WorkloadService.Commands.SolicitationTask.ApproveSubmission,
+      WorkloadService.Commands.SolicitationTask.CompleteTask
+    ],
+    to: WorkloadService.Aggregates.SolicitationTask,
+    identity: :id
+  )
+end
+
+defmodule WorkloadService.CommandedApp do
+  use Commanded.Application,
+    otp_app: :workload_service
+
+  router(WorkloadService.Router)
+end

lib/workload_service/commands/quote_task.ex

@@ -0,0 +1,53 @@
+defmodule WorkloadService.Commands.QuoteTask do
+  @moduledoc """
+  Quote task commands.
+  """
+
+  defmodule CreateTask do
+    @moduledoc """
+    Command to create a new quote task.
+    """
+    @derive Jason.Encoder
+    defstruct [:id, :application_id, :provider_id, :provider_name, :task_info, :attachments]
+
+    def new(attrs) do
+      struct(__MODULE__, attrs)
+    end
+  end
+
+  defmodule SubmitResponse do
+    @moduledoc """
+    Command to submit response for a quote task.
+    """
+    @derive Jason.Encoder
+    defstruct [:id, :submission, :attachments]
+
+    def new(attrs) do
+      struct(__MODULE__, attrs)
+    end
+  end
+
+  defmodule ApproveSubmission do
+    @moduledoc """
+    Command to approve submission for a quote task.
+    """
+    @derive Jason.Encoder
+    defstruct [:id]
+
+    def new(attrs) do
+      struct(__MODULE__, attrs)
+    end
+  end
+
+  defmodule CompleteTask do
+    @moduledoc """
+    Command to complete a quote task.
+    """
+    @derive Jason.Encoder
+    defstruct [:id, :completed_by]
+
+    def new(attrs) do
+      struct(__MODULE__, attrs)
+    end
+  end
+end

lib/workload_service/commands/solicitation_task.ex

@@ -0,0 +1,53 @@
+defmodule WorkloadService.Commands.SolicitationTask do
+  @moduledoc """
+  Solicitation task commands.
+  """
+
+  defmodule CreateTask do
+    @moduledoc """
+    Command to create a new solicitation task.
+    """
+    @derive Jason.Encoder
+    defstruct [:id, :application_id, :provider_id, :provider_name, :task_info, :attachments]
+
+    def new(attrs) do
+      struct(__MODULE__, attrs)
+    end
+  end
+
+  defmodule SubmitResponse do
+    @moduledoc """
+    Command to submit response for a solicitation task.
+    """
+    @derive Jason.Encoder
+    defstruct [:id, :submission, :attachments]
+
+    def new(attrs) do
+      struct(__MODULE__, attrs)
+    end
+  end
+
+  defmodule ApproveSubmission do
+    @moduledoc """
+    Command to approve submission for a solicitation task.
+    """
+    @derive Jason.Encoder
+    defstruct [:id]
+
+    def new(attrs) do
+      struct(__MODULE__, attrs)
+    end
+  end
+
+  defmodule CompleteTask do
+    @moduledoc """
+    Command to complete a solicitation task.
+    """
+    @derive Jason.Encoder
+    defstruct [:id, :completed_by]
+
+    def new(attrs) do
+      struct(__MODULE__, attrs)
+    end
+  end
+end

lib/workload_service/consumers/quote_requested_consumer.ex

@@ -0,0 +1,93 @@
+defmodule WorkloadService.Consumers.QuoteRequestedConsumer do
+  use GenServer
+  require Logger
+
+  alias WorkloadService.CommandedApp
+  alias WorkloadService.Commands.QuoteTask
+
+  @exchange "workload_service.events.quote_requested"
+  @queue "workload_service.quote_requested"
+  @routing_key "quote.requested"
+
+  def start_link(opts \\ []) do
+    GenServer.start_link(__MODULE__, opts, name: __MODULE__)
+  end
+
+  def init(_opts) do
+    {:ok, conn} = AMQP.Connection.open(amqp_url())
+    {:ok, channel} = AMQP.Channel.open(conn)
+
+    AMQP.Queue.declare(channel, @queue, durable: true)
+    AMQP.Queue.bind(channel, @queue, @exchange, routing_key: @routing_key)
+    {:ok, _tag} = AMQP.Basic.consume(channel, @queue)
+
+    Logger.info("QuoteRequestedConsumer started, listening on #{@queue}")
+
+    {:ok, %{channel: channel}}
+  end
+
+  def handle_info({:basic_consume_ok, _}, state), do: {:noreply, state}
+  def handle_info({:basic_cancel, _}, state), do: {:stop, :normal, state}
+  def handle_info({:basic_cancel_ok, _}, state), do: {:noreply, state}
+
+  def handle_info({:basic_deliver, payload, meta}, state) do
+    case process(payload) do
+      :ok ->
+        AMQP.Basic.ack(state.channel, meta.delivery_tag)
+
+      {:error, reason} ->
+        Logger.error("QuoteRequestedConsumer: failed to process: #{inspect(reason)}")
+        AMQP.Basic.reject(state.channel, meta.delivery_tag, requeue: false)
+    end
+
+    {:noreply, state}
+  end
+
+  defp process(payload) do
+    with {:ok, event} <- Jason.decode(payload),
+         :ok <- handle_event(event) do
+      :ok
+    end
+  end
+
+  defp handle_event(%{
+    "application_id" => application_id,
+    "org_id" => org_id,
+    "provider_id" => provider_id,
+    "policy_type" => policy_type
+  } = event) do
+    task_id = WorkloadService.Aggregates.TaskId.new(org_id, "quote", Ecto.UUID.generate())
+
+    command = %QuoteTask.CreateTask{
+      id: task_id,
+      application_id: application_id,
+      provider_id: provider_id,
+      provider_name: Map.get(event, "provider_name", ""),
+      task_info: %{
+        "policy_type" => policy_type,
+        "provider_email" => Map.get(event, "provider_email"),
+        "applicant_info" => Map.get(event, "applicant_info", %{}),
+        "car_details" => Map.get(event, "car_details", %{}),
+        "building_details" => Map.get(event, "building_details", %{}),
+        "life_details" => Map.get(event, "life_details", %{})
+      }
+    }
+
+    case CommandedApp.dispatch(command) do
+      :ok ->
+        Logger.info("QuoteRequestedConsumer: created task #{task_id}")
+        :ok
+
+      {:error, reason} ->
+        {:error, reason}
+    end
+  end
+
+  defp handle_event(event) do
+    {:error, {:invalid_event, event}}
+  end
+
+  defp amqp_url do
+    Application.get_env(:workload_service, :amqp_url, "amqp://guest:guest@localhost:5672")
+  end
+end

lib/workload_service/consumers/solicitation_requested_consumer.ex

@@ -0,0 +1,138 @@
+defmodule WorkloadService.Consumers.SolicitationRequestedConsumer do
+  use GenServer
+  require Logger
+
+  alias WorkloadService.CommandedApp
+  alias WorkloadService.Commands.SolicitationTask
+
+  @exchange "workload_service.events.solicitation_requested"
+  @queue "workload_service.solicitation_requested"
+  @routing_key "solicitation.requested"
+
+  @provider_service_url "http://localhost:4002"
+  @solicitation_service_url "http://localhost:8081"
+
+  def start_link(opts \\ []) do
+    GenServer.start_link(__MODULE__, opts, name: __MODULE__)
+  end
+
+  def init(_opts) do
+    {:ok, conn} = AMQP.Connection.open(amqp_url())
+    {:ok, channel} = AMQP.Channel.open(conn)
+
+    AMQP.Queue.declare(channel, @queue, durable: true)
+    AMQP.Queue.bind(channel, @queue, @exchange, routing_key: @routing_key)
+    {:ok, _tag} = AMQP.Basic.consume(channel, @queue)
+
+    Logger.info("SolicitationRequestedConsumer started, listening on #{@queue}")
+
+    {:ok, %{channel: channel}}
+  end
+
+  def handle_info({:basic_consume_ok, _}, state), do: {:noreply, state}
+  def handle_info({:basic_cancel, _}, state), do: {:stop, :normal, state}
+  def handle_info({:basic_cancel_ok, _}, state), do: {:noreply, state}
+
+  def handle_info({:basic_deliver, payload, meta}, state) do
+    case process(payload) do
+      :ok ->
+        AMQP.Basic.ack(state.channel, meta.delivery_tag)
+
+      {:error, reason} ->
+        Logger.error("SolicitationRequestedConsumer: failed to process: #{inspect(reason)}")
+        AMQP.Basic.reject(state.channel, meta.delivery_tag, requeue: false)
+    end
+
+    {:noreply, state}
+  end
+
+  defp process(payload) do
+    with {:ok, event} <- Jason.decode(payload),
+         {:ok, provider} <- get_provider(event["provider_id"]),
+         {:ok, template} <- get_active_template(provider, event["policy_type"]),
+         {:ok, result} <- generate_solicitation(event, template),
+         :ok <- create_task(event, result) do
+      :ok
+    end
+  end
+
+  defp get_provider(provider_id) do
+    url = "#{@provider_service_url}/api/v1/providers/#{provider_id}"
+
+    case Req.get(url) do
+      {:ok, %{status: 200, body: %{"data" => provider}}} -> {:ok, provider}
+      {:ok, %{status: 404}} -> {:error, :provider_not_found}
+      {:error, reason} -> {:error, reason}
+    end
+  end
+
+  defp get_active_template(provider, policy_type) do
+    templates = get_in(provider, ["templates", policy_type]) || []
+    default_id = get_in(provider, ["default_templates", policy_type])
+
+    template =
+      if default_id do
+        Enum.find(templates, &(&1["template_id"] == default_id))
+      else
+        Enum.find(templates, &(&1["active"] == true))
+      end
+
+    case template do
+      nil -> {:error, :no_active_template}
+      t -> {:ok, t}
+    end
+  end
+
+  defp generate_solicitation(event, template) do
+    url = "#{@solicitation_service_url}/api/solicitations/generate"
+
+    body = %{
+      org_id: event["org_id"],
+      id: event["id"],
+      template_document_url: template["document_url"],
+      fields: event["fields"] || %{}
+    }
+
+    case Req.post(url, json: body) do
+      {:ok, %{status: 200, body: result}} ->
+        {:ok, result}
+
+      {:ok, %{status: status, body: body}} ->
+        {:error, "solicitation_service returned #{status}: #{inspect(body)}"}
+
+      {:error, reason} ->
+        {:error, reason}
+    end
+  end
+
+  defp create_task(event, result) do
+    org_id = event["org_id"]
+    task_id = WorkloadService.Aggregates.TaskId.new(org_id, "solicitation", Ecto.UUID.generate())
+
+    command = %SolicitationTask.CreateTask{
+      id: task_id,
+      application_id: event["id"],
+      provider_id: event["provider_id"],
+      provider_name: event["provider_name"] || "",
+      task_info: %{
+        "quote_id" => event["quote_id"],
+        "plan_id" => event["plan_id"],
+        "document_url" => result["document_url"]
+      },
+      attachments: [result["document_url"]] |> Enum.filter(& &1)
+    }
+
+    case CommandedApp.dispatch(command) do
+      :ok ->
+        Logger.info("SolicitationRequestedConsumer: created task #{task_id}")
+        :ok
+
+      {:error, reason} ->
+        {:error, reason}
+    end
+  end
+
+  defp amqp_url do
+    Application.get_env(:workload_service, :amqp_url, "amqp://guest:guest@localhost:5672")
+  end
+end

lib/workload_service/event_store.ex

@@ -0,0 +1,3 @@
+defmodule WorkloadService.EventStore do
+  use EventStore, otp_app: :workload_service
+end

lib/workload_service/events/task.ex

@@ -0,0 +1,47 @@
+defmodule WorkloadService.Events do
+  @moduledoc """
+  All domain events for the workload service.
+  """
+
+  defmodule TaskCreated do
+    @moduledoc """
+    Emitted when a new task is created (quote or solicitation).
+    ID format: "org_id:type:task_id" (e.g., "test:quote:uuid") - TaskId struct
+    """
+    @derive Jason.Encoder
+    defstruct [:id, :application_id, :provider_id, :provider_name, :task_info, :attachments]
+  end
+
+  defmodule SubmissionUpdated do
+    @moduledoc """
+    Emitted when submission is updated (user provides response data).
+    """
+    @derive Jason.Encoder
+    defstruct [:id, :submission, :attachments]
+  end
+
+  defmodule SubmissionApproved do
+    @moduledoc """
+    Emitted when submission is approved and ready to send.
+    """
+    @derive Jason.Encoder
+    defstruct [:id]
+  end
+
+  defmodule TaskCompleted do
+    @moduledoc """
+    Emitted when task is completed and sent to policy-service.
+    Triggers RabbitMQ publish.
+    """
+    @derive Jason.Encoder
+    defstruct [:id, :completed_by]
+  end
+
+  defmodule QuoteAccepted do
+    @moduledoc """
+    Emitted when a quote is accepted - triggers solicitation generation.
+    """
+    @derive Jason.Encoder
+    defstruct [:id, :provider_id, :quote_id, :plan_id]
+  end
+end

lib/workload_service/message_bus.ex

@@ -0,0 +1,86 @@
+defmodule WorkloadService.MessageBus do
+  @moduledoc false
+
+  use GenServer
+  require Logger
+
+  alias AMQP.{Connection, Channel, Exchange}
+
+  @reconnect_interval 5_000
+
+  def start_link(opts \\ []) do
+    GenServer.start_link(__MODULE__, opts, name: __MODULE__)
+  end
+
+  def publish(routing_key, payload) when is_binary(payload) do
+    GenServer.cast(__MODULE__, {:publish, routing_key, payload})
+  end
+
+  def publish(routing_key, payload) do
+    GenServer.cast(__MODULE__, {:publish, routing_key, Jason.encode!(payload)})
+  end
+
+  @impl true
+  def init(_opts) do
+    send(self(), :connect)
+    {:ok, %{channel: nil, connection: nil}}
+  end
+
+  @impl true
+  def handle_info(:connect, _state) do
+    case connect() do
+      {:ok, connection, channel} ->
+        {:noreply, %{connection: connection, channel: channel}}
+
+      {:error, reason} ->
+        Logger.error("Failed to connect to RabbitMQ: #{inspect(reason)}")
+        schedule_reconnect()
+        {:noreply, %{channel: nil, connection: nil}}
+    end
+  end
+
+  @impl true
+  def handle_info({:DOWN, _, :process, _pid, reason}, _state) do
+    Logger.error("RabbitMQ connection lost: #{inspect(reason)}")
+    schedule_reconnect()
+    {:noreply, %{channel: nil, connection: nil}}
+  end
+
+  @impl true
+  def handle_cast({:publish, _routing_key, _payload}, %{channel: nil} = state) do
+    Logger.warning("RabbitMQ not connected, message dropped")
+    {:noreply, state}
+  end
+
+  @impl true
+  def handle_cast({:publish, routing_key, payload}, %{channel: channel} = state) do
+    Exchange.direct(channel, exchange_name(), durable: true)
+
+    AMQP.Basic.publish(channel, exchange_name(), routing_key, payload,
+      content_type: "application/json",
+      persistent: true
+    )
+
+    {:noreply, state}
+  end
+
+  defp connect do
+    config = Application.get_env(:workload_service, WorkloadService.MessageBus, [])
+
+    with {:ok, connection} <- Connection.open(config),
+         {:ok, channel} <- Channel.open(connection) do
+      Exchange.declare(channel, exchange_name(), :direct, durable: true)
+      Process.monitor(connection.pid)
+      {:ok, connection, channel}
+    end
+  end
+
+  defp exchange_name do
+    Application.get_env(:workload_service, WorkloadService.MessageBus, [])
+    |> Keyword.get(:exchange, "workload_service.events")
+  end
+
+  defp schedule_reconnect do
+    Process.send_after(self(), :connect, @reconnect_interval)
+  end
+end

lib/workload_service/projections/task.ex

@@ -0,0 +1,61 @@
+defmodule WorkloadService.Projections.Task do
+  @moduledoc """
+  Ecto schema for the task read model.
+  ID format: "org_id:type:task_id" (e.g., "test:quote:uuid")
+  """
+
+  use Ecto.Schema
+  import Ecto.Changeset
+
+  @derive {Flop.Schema,
+           filterable: [:status, :org_id, :application_id, :provider_id],
+           sortable: [:inserted_at, :updated_at, :status]}
+
+  @primary_key {:id, :string, []}
+  schema "tasks" do
+    field(:org_id, :string)
+    field(:application_id, :string)
+    field(:provider_id, :string)
+    field(:provider_name, :string)
+    field(:task_info, :map)
+    field(:submission, :map)
+    field(:attachments, {:array, :string})
+    field(:status, :string)
+    field(:version, :integer, default: 1)
+
+    timestamps(type: :utc_datetime)
+  end
+
+  @statuses ["created", "draft", "approved", "completed"]
+
+  def status_changeset(task, attrs) do
+    task
+    |> cast(attrs, [:status, :task_info, :submission, :attachments, :updated_at])
+    |> validate_inclusion(:status, @statuses)
+  end
+
+  def create_changeset(task, attrs) do
+    task
+    |> cast(attrs, [
+      :id,
+      :org_id,
+      :application_id,
+      :provider_id,
+      :provider_name,
+      :task_info,
+      :submission,
+      :attachments,
+      :status,
+      :version
+    ])
+    |> validate_required([
+      :id,
+      :org_id,
+      :application_id,
+      :provider_id,
+      :provider_name,
+      :status
+    ])
+    |> validate_inclusion(:status, @statuses)
+  end
+end

lib/workload_service/projectors/task_projector.ex

@@ -0,0 +1,61 @@
+defmodule WorkloadService.Projectors.TaskProjector do
+  use Commanded.Projections.Ecto,
+    application: WorkloadService.CommandedApp,
+    repo: WorkloadService.Repo,
+    name: "TaskProjector"
+
+  alias WorkloadService.Aggregates.TaskId
+  alias WorkloadService.Events
+  alias WorkloadService.Projections.Task
+  import Ecto.Query
+
+  project(%Events.TaskCreated{} = e, _meta, fn multi ->
+    task_id = TaskId.parse!(to_string(e.id))
+    %{org_id: org_id} = task_id
+
+    Ecto.Multi.insert(multi, :task, %Task{
+      id: to_string(e.id),
+      org_id: org_id,
+      application_id: e.application_id,
+      provider_id: e.provider_id,
+      provider_name: e.provider_name,
+      task_info: e.task_info,
+      attachments: e.attachments || [],
+      status: "created"
+    })
+  end)
+
+  project(%Events.SubmissionUpdated{} = e, _meta, fn multi ->
+    multi
+    |> Ecto.Multi.run(:fetch, fn repo, _ ->
+      {:ok, repo.get(Task, to_string(e.id))}
+    end)
+    |> Ecto.Multi.update(:task, fn %{fetch: task} ->
+      Ecto.Changeset.change(task, %{
+        status: "draft",
+        submission: e.submission,
+        attachments: e.attachments || []
+      })
+    end)
+  end)
+
+  project(%Events.SubmissionApproved{} = e, _meta, fn multi ->
+    multi
+    |> Ecto.Multi.run(:fetch, fn repo, _ ->
+      {:ok, repo.get(Task, to_string(e.id))}
+    end)
+    |> Ecto.Multi.update(:task, fn %{fetch: task} ->
+      Ecto.Changeset.change(task, %{status: "approved"})
+    end)
+  end)
+
+  project(%Events.TaskCompleted{} = e, _meta, fn multi ->
+    multi
+    |> Ecto.Multi.run(:fetch, fn repo, _ ->
+      {:ok, repo.get(Task, to_string(e.id))}
+    end)
+    |> Ecto.Multi.update(:task, fn %{fetch: task} ->
+      Ecto.Changeset.change(task, %{status: "completed"})
+    end)
+  end)
+end

lib/workload_service/release.ex

@@ -0,0 +1,37 @@
+defmodule WorkloadService.Release do
+  @moduledoc """
+  Used for executing DB release tasks when run in production without Mix
+  installed.
+  """
+  @app :workload_service
+
+  def migrate do
+    load_app()
+    init_event_store()
+
+    for repo <- repos() do
+      {:ok, _, _} = Ecto.Migrator.with_repo(repo, &Ecto.Migrator.run(&1, :up, all: true))
+    end
+  end
+
+  def rollback(repo, version) do
+    load_app()
+    {:ok, _, _} = Ecto.Migrator.with_repo(repo, &Ecto.Migrator.run(&1, :down, to: version))
+  end
+
+  defp repos do
+    Application.fetch_env!(@app, :ecto_repos)
+  end
+
+  defp load_app do
+    Application.ensure_all_started(:ssl)
+    Application.ensure_all_started(:postgrex)
+    Application.ensure_loaded(@app)
+  end
+
+  def init_event_store do
+    config = WorkloadService.EventStore.config()
+
+    :ok = EventStore.Tasks.Init.exec(config, [])
+  end
+end

lib/workload_service/repo.ex

@@ -0,0 +1,7 @@
+defmodule WorkloadService.Repo do
+  @moduledoc false
+
+  use Ecto.Repo,
+    otp_app: :workload_service,
+    adapter: Ecto.Adapters.Postgres
+end

lib/workload_service/workload/queries.ex

@@ -0,0 +1,38 @@
+defmodule WorkloadService.Workload.Queries do
+  @moduledoc """
+  Database query functions for tasks.
+  """
+
+  import Ecto.Query
+  alias WorkloadService.Projections.Task
+  alias WorkloadService.Repo
+
+  def list_tasks(params \\ %{}) do
+    base_query()
+    |> Flop.validate_and_run(params, for: Task)
+  end
+
+  def list_tasks_by_org(org_id, params \\ %{}) do
+    base_query()
+    |> where(org_id: ^org_id)
+    |> Flop.validate_and_run(params, for: Task)
+  end
+
+  def get_task_by_id(id) do
+    case Repo.get(Task, id) do
+      nil -> {:error, :not_found}
+      task -> {:ok, task}
+    end
+  end
+
+  def get_task_by_org_and_application(org_id, application_id) do
+    case Repo.get_by(Task, org_id: org_id, application_id: application_id) do
+      nil -> {:error, :not_found}
+      task -> {:ok, task}
+    end
+  end
+
+  defp base_query do
+    from(t in Task, as: :task)
+  end
+end