Sequential selection WIP

lib/designator/selection.ex

@@ -74,12 +74,20 @@ defmodule Designator.Selection do
   end
 
   @spec convert_to_streams([SubjectSetCache.t], Workflow.t) :: [SubjectStream.t]
-  def convert_to_streams(subject_sets, _workflow) do
+  def convert_to_streams(subject_sets, workflow) do
     Enum.map(subject_sets, fn subject_set ->
-      Designator.SubjectStream.build(subject_set)
+      Designator.SubjectStream.build(subject_set, prepare_iterator(workflow))
     end)
   end
 
+  defp prepare_iterator(workflow) do
+    if workflow.prioritized do
+      Designator.SubjectSetIterators.Sequentially
+    else
+      Designator.SubjectSetIterators.Randomly
+    end
+  end
+
   defp deduplicate(stream) do
     Stream.uniq(stream)
   end

lib/designator/random_stream.ex → lib/designator/subject_set_iterators/randomly.ex

@@ -1,8 +1,8 @@
-defmodule Designator.RandomStream do
+defmodule Designator.SubjectSetIterators.Randomly do
   alias Designator.Random
 
-  @spec shuffle(Enumerable.t) :: Enumerable.t
-  def shuffle(enum) do
+  @spec apply_to(Enumerable.t) :: Enumerable.t
+  def apply_to(enum) do
     Stream.unfold({enum, MapSet.new}, fn {enum, drawn} ->
       if size(enum) <= MapSet.size(drawn) do
         nil

lib/designator/subject_set_iterators/sequentially.ex

@@ -0,0 +1,15 @@
+defmodule Designator.SubjectSetIterators.Sequentially do
+
+  @spec apply_to(Enumerable.t) :: Enumerable.t
+  def apply_to(enum) do
+    Stream.with_index(enum) |> Stream.map(fn {elem,index} -> {index, elem} end)
+  end
+
+  defp size(enum = %Array{}) do
+    Array.size(enum)
+  end
+
+  defp size(enum) do
+    Enum.count(enum)
+  end
+end

lib/designator/subject_stream.ex

@@ -1,15 +1,15 @@
 defmodule Designator.SubjectStream do
   defstruct [:subject_set_id, :stream, :amount, :chance]
 
-  def build(%{subject_set_id: subject_set_id, subject_ids: subject_ids}) do
+  def build(%{subject_set_id: subject_set_id, subject_ids: subject_ids}, subject_set_iterator) do
     amount = get_amount(subject_ids)
-    %Designator.SubjectStream{subject_set_id: subject_set_id, stream: build_stream(subject_ids), amount: amount, chance: amount}
+    %Designator.SubjectStream{subject_set_id: subject_set_id, stream: build_stream(subject_ids, subject_set_iterator), amount: amount, chance: amount}
   end
 
   ###
 
-  defp build_stream(subject_ids) do
-    Designator.RandomStream.shuffle(subject_ids) |> Stream.map(fn {_idx, elm} -> elm end)
+  defp build_stream(subject_ids, subject_set_iterator) do
+    subject_set_iterator.apply_to(subject_ids) |> Stream.map(fn {_idx, elm} -> elm end)
   end
 
   def get_amount(%Array{} = subject_ids) do

lib/designator/workflow_cache.ex

@@ -18,7 +18,7 @@ defmodule Designator.WorkflowCache do
 
   ### Public API
 
-  defstruct [:id, :subject_set_ids, :configuration]
+  defstruct [:id, :subject_set_ids, :configuration, :prioritized]
 
   def status do
     :workflow_cache
@@ -60,13 +60,15 @@ defmodule Designator.WorkflowCache do
         %__MODULE__{
           id: workflow_id,
           subject_set_ids: [],
-          configuration: %{}
+          configuration: %{},
+          prioritized: false
         }
       workflow ->
         %__MODULE__{
           id: workflow_id,
           subject_set_ids: Designator.Workflow.subject_set_ids(workflow_id),
-          configuration: workflow.configuration
+          configuration: workflow.configuration,
+          prioritized: workflow.prioritized
        }
     end
   end

test/designator/selection_test.exs

@@ -45,6 +45,15 @@ defmodule Designator.SelectionTest do
     assert Selection.select(338, 1, [limit: 4]) == [4, 2, 3, 1]
   end
 
+  test "sequential selection for normal sets" do
+    Designator.Random.seed({123, 100020, 345345})
+    Designator.WorkflowCache.set(338, %{configuration: %{}, prioritized: true, subject_set_ids: [1000]})
+    Designator.UserCache.set({338, 1}, %{seen_ids: MapSet.new, recently_selected_ids: MapSet.new, configuration: %{}})
+    SubjectSetCache.set({338, 1000}, %SubjectSetCache{workflow_id: 338, subject_set_id: 1000, subject_ids: Array.from_list([98, 99, 10])})
+
+    assert Selection.select(338, 1, [limit: 6]) == [98, 99, 10]
+  end
+
   test "weighed selection for normal sets" do
     Designator.Random.seed({123, 100020, 345345})
     Designator.WorkflowCache.set(338, %{configuration: %{"subject_set_weights" => %{"1000" => 1, "1001" => 99, "1002" => 9.9, "1003" => 0.1}},

test/designator/subject_set_iterators/randomly_test.exs

@@ -0,0 +1,13 @@
+defmodule Designator.SubjectSetIterators.RandomlyTest do
+  use ExUnit.Case
+
+  import Designator.SubjectSetIterators.Randomly
+
+  test "empty enum returns nothing" do
+    assert ([] |> apply_to |> Stream.take(5) |> Enum.sort) == []
+  end
+
+  test "returns data" do
+    assert (1..5 |> apply_to |> Stream.take(5) |> Enum.sort) == [{0, 1}, {1, 2}, {2, 3}, {3, 4}, {4, 5}]
+  end
+end

test/designator/subject_set_iterators/sequentially_test.exs

@@ -0,0 +1,13 @@
+defmodule Designator.SubjectSetIterators.SequentiallyTest do
+  use ExUnit.Case
+
+  import Designator.SubjectSetIterators.Sequentially
+
+  test "empty enum returns nothing" do
+    assert ([] |> apply_to |> Stream.take(5) |> Enum.sort) == []
+  end
+
+  test "returns data" do
+    assert (1..5 |> apply_to |> Stream.take(5)) |> Enum.into([]) == [{0, 1}, {1, 2}, {2, 3}, {3, 4}, {4, 5}]
+  end
+end

test/models/workflow_test.exs

@@ -14,21 +14,21 @@ defmodule Designator.WorkflowTest do
     test "returns subject ids" do
       Ecto.Adapters.SQL.query!(Designator.Repo, "INSERT INTO workflows (id, created_at, updated_at) VALUES (1, NOW(), NOW())")
       Ecto.Adapters.SQL.query!(Designator.Repo, "INSERT INTO subject_sets_workflows (workflow_id, subject_set_id) VALUES (1, 1)")
-      Ecto.Adapters.SQL.query!(Designator.Repo, "INSERT INTO set_member_subjects (subject_set_id, subject_id, random, created_at, updated_at) VALUES
-        (1, 1, 0.5, NOW(), NOW()),
-        (1, 2, 0.5, NOW(), NOW()),
-        (1, 3, 0.5, NOW(), NOW())")
+      Ecto.Adapters.SQL.query!(Designator.Repo, "INSERT INTO set_member_subjects (subject_set_id, subject_id, priority, random, created_at, updated_at) VALUES
+        (1, 1, 3, 0.5, NOW(), NOW()),
+        (1, 2, 2, 0.5, NOW(), NOW()),
+        (1, 3, 1, 0.5, NOW(), NOW())")
 
       assert Designator.Workflow.subject_ids(1, 1) |> Enum.sort == [1,2,3]
     end
 
     test "does not return retired subjects" do
       Ecto.Adapters.SQL.query!(Designator.Repo, "INSERT INTO workflows (id, created_at, updated_at) VALUES (1, NOW(), NOW())")
       Ecto.Adapters.SQL.query!(Designator.Repo, "INSERT INTO subject_sets_workflows (workflow_id, subject_set_id) VALUES (1, 1)")
-      Ecto.Adapters.SQL.query!(Designator.Repo, "INSERT INTO set_member_subjects (subject_set_id, subject_id, random, created_at, updated_at) VALUES
-      (1, 1, 0.5, NOW(), NOW()),
-      (1, 2, 0.5, NOW(), NOW()),
-      (1, 3, 0.5, NOW(), NOW())")
+      Ecto.Adapters.SQL.query!(Designator.Repo, "INSERT INTO set_member_subjects (subject_set_id, subject_id, priority, random, created_at, updated_at) VALUES
+      (1, 1, 3, 0.5, NOW(), NOW()),
+      (1, 2, 2, 0.5, NOW(), NOW()),
+      (1, 3, 1, 0.5, NOW(), NOW())")
       Ecto.Adapters.SQL.query!(Designator.Repo, "INSERT INTO subject_workflow_counts (workflow_id, subject_id, retired_at) VALUES
       (1, 1, NOW())")
 
@@ -40,15 +40,25 @@ defmodule Designator.WorkflowTest do
       (1, NOW(), NOW()),
       (2, NOW(), NOW())")
       Ecto.Adapters.SQL.query!(Designator.Repo, "INSERT INTO subject_sets_workflows (workflow_id, subject_set_id) VALUES (1, 1)")
-      Ecto.Adapters.SQL.query!(Designator.Repo, "INSERT INTO set_member_subjects (subject_set_id, subject_id, random, created_at, updated_at) VALUES
-      (1, 1, 0.5, NOW(), NOW()),
-      (1, 2, 0.5, NOW(), NOW()),
-      (1, 3, 0.5, NOW(), NOW())")
+      Ecto.Adapters.SQL.query!(Designator.Repo, "INSERT INTO set_member_subjects (subject_set_id, subject_id, priority, random, created_at, updated_at) VALUES
+      (1, 1, 3, 0.5, NOW(), NOW()),
+      (1, 2, 2, 0.5, NOW(), NOW()),
+      (1, 3, 1, 0.5, NOW(), NOW())")
       Ecto.Adapters.SQL.query!(Designator.Repo, "INSERT INTO subject_workflow_counts (workflow_id, subject_id, retired_at) VALUES
       (1, 1, NULL),
       (2, 1, NULL)")
 
       assert Designator.Workflow.subject_ids(1, 1) |> Enum.sort == [1,2,3]
     end
+
+    test "maintains priority order" do
+      Ecto.Adapters.SQL.query!(Designator.Repo, "INSERT INTO workflows (id, created_at, updated_at) VALUES (1, NOW(), NOW())")
+      Ecto.Adapters.SQL.query!(Designator.Repo, "INSERT INTO subject_sets_workflows (workflow_id, subject_set_id) VALUES (1, 1)")
+      Ecto.Adapters.SQL.query!(Designator.Repo, "INSERT INTO set_member_subjects (subject_set_id, subject_id, priority, random, created_at, updated_at) VALUES
+        (1, 1, 3, 0.5, NOW(), NOW()),
+        (1, 2, 2, 0.5, NOW(), NOW()),
+        (1, 3, 1, 0.5, NOW(), NOW())")
+      assert Designator.Workflow.subject_ids(1, 1) == [3,2,1]
+    end
   end
 end

web/models/workflow.ex

@@ -7,6 +7,7 @@ defmodule Designator.Workflow do
   schema "workflows" do
     field :project_id, :integer
     field :configuration, :map
+    field :prioritized, :boolean
 
     timestamps inserted_at: :created_at
   end
@@ -27,8 +28,11 @@ defmodule Designator.Workflow do
     query = from sms in "set_member_subjects",
       left_join: swc in "subject_workflow_counts", on: (sms.subject_id == swc.subject_id and swc.workflow_id == ^workflow_id),
       where: sms.subject_set_id == ^subject_set_id and is_nil(swc.retired_at),
-      select: sms.subject_id
+      select: {sms.subject_id, sms.priority}
+
     Designator.Repo.all(query)
+    |> List.keysort(1)
+    |> Enum.map(fn {k, v}->k end)
   end
 
   def changeset(struct, params \\ %{}) do