+119 -36

apps/sower/lib/sower/orchestration.ex

reviewed

··· 1081 1081 1082 1082 @doc """ 1083 1083 Upserts an agent_seed_generation from report data. 1084 1084 - Uses the unique constraint on (agent_id, seed_id) for conflict resolution. 1084 1084 + Uses lookup semantics on (agent_id, seed_id), inserting missing rows and 1085 1085 + updating existing rows. 1085 1086 1086 1087 ## Parameters 1087 1088 - agent_id: The agent's ID ··· 1102 1103 created_at_generation: attrs.created_at_generation 1103 1104 } 1104 1105 1105 1105 - %AgentSeedGeneration{} 1106 1106 - |> AgentSeedGeneration.changeset(changeset_attrs) 1107 1107 - |> Repo.insert( 1108 1108 - on_conflict: [ 1109 1109 - set: [ 1106 1106 + case Repo.get_by(AgentSeedGeneration, agent_id: agent_id, seed_id: seed_id) do 1107 1107 + nil -> 1108 1108 + %AgentSeedGeneration{} 1109 1109 + |> AgentSeedGeneration.changeset(changeset_attrs) 1110 1110 + |> Repo.insert() 1111 1111 + 1112 1112 + %AgentSeedGeneration{} = existing -> 1113 1113 + update_attrs = %{ 1110 1114 profile_id: profile_id, 1111 1115 generation_number: attrs.generation_number, 1112 1116 is_current: attrs.is_current, 1117 1117 + created_at_generation: attrs.created_at_generation, 1113 1118 updated_at: now 1114 1114 - ] 1115 1115 - ], 1116 1116 - conflict_target: [:agent_id, :seed_id] 1117 1117 - ) 1119 1119 + } 1120 1120 + 1121 1121 + if generation_row_changed?(existing, update_attrs) do 1122 1122 + existing 1123 1123 + |> AgentSeedGeneration.changeset(update_attrs) 1124 1124 + |> Repo.update() 1125 1125 + else 1126 1126 + {:ok, existing} 1127 1127 + end 1128 1128 + end 1118 1129 end 1119 1130 1120 1131 @doc """ ··· 1135 1146 Repo.transaction(fn -> 1136 1147 for profile <- report.profiles do 1137 1148 nix_profile = NixProfile.find_or_create!(profile.profile_path) 1149 1149 + rows = resolve_profile_generation_rows(agent, profile) 1150 1150 + sync_profile_generation_rows(agent, nix_profile, rows) 1151 1151 + end 1138 1152 1139 1139 - # Collect seed_ids we're upserting for this profile 1140 1140 - upserted_seed_ids = 1141 1141 - for gen <- profile.generations, reduce: [] do 1142 1142 - acc -> 1153 1153 + :ok 1154 1154 + end) 1155 1155 + end 1156 1156 + 1157 1157 + defp resolve_profile_generation_rows(%Agent{} = agent, profile) do 1158 1158 + artifacts = 1159 1159 + profile.generations 1160 1160 + |> Enum.map(& &1.path) 1161 1161 + |> Enum.uniq() 1162 1162 + 1163 1163 + seeds_by_artifact = 1164 1164 + from(s in Seed, where: s.artifact in ^artifacts) 1165 1165 + |> Repo.all() 1166 1166 + |> Map.new(&{&1.artifact, &1}) 1167 1167 + 1168 1168 + {rows, _seeds_by_artifact} = 1169 1169 + Enum.reduce(profile.generations, {[], seeds_by_artifact}, fn gen, {rows, seeds} -> 1170 1170 + {seed, seeds} = 1171 1171 + case Map.get(seeds, gen.path) do 1172 1172 + nil -> 1143 1173 case Seed.find_or_register(agent, gen, profile) do 1144 1174 {:ok, seed} -> 1145 1145 - upsert_agent_seed_generation(agent, nix_profile, seed, gen) 1146 1146 - [seed.id | acc] 1175 1175 + {seed, Map.put(seeds, gen.path, seed)} 1147 1176 1148 1177 {:error, error} -> 1149 1178 Logger.warning( ··· 1152 1181 error: error 1153 1182 ) 1154 1183 1155 1155 - acc 1184 1184 + {nil, seeds} 1156 1185 end 1186 1186 + 1187 1187 + seed -> 1188 1188 + {seed, seeds} 1157 1189 end 1158 1190 1159 1159 - # Delete agent_seed_generations for this profile that are no longer in the report 1160 1160 - delete_stale_agent_seed_generations(agent.id, nix_profile.id, upserted_seed_ids) 1161 1161 - end 1191 1191 + case {seed, parse_generation_created(gen.created)} do 1192 1192 + {nil, _} -> 1193 1193 + {rows, seeds} 1194 1194 + 1195 1195 + {_, :error} -> 1196 1196 + Logger.warning( 1197 1197 + msg: "Failed to parse generation created timestamp", 1198 1198 + artifact: gen.path, 1199 1199 + created: gen.created 1200 1200 + ) 1201 1201 + 1202 1202 + {rows, seeds} 1203 1203 + 1204 1204 + {%Seed{id: seed_id}, {:ok, created_at}} -> 1205 1205 + row = %{ 1206 1206 + seed_id: seed_id, 1207 1207 + generation_number: gen.generation_number, 1208 1208 + is_current: gen.is_current, 1209 1209 + created_at_generation: created_at 1210 1210 + } 1211 1211 + 1212 1212 + {[row | rows], seeds} 1213 1213 + end 1214 1214 + end) 1215 1215 + 1216 1216 + rows 1217 1217 + |> Enum.reverse() 1218 1218 + |> normalize_current_generation_rows() 1219 1219 + |> Enum.reverse() 1220 1220 + |> Enum.uniq_by(& &1.seed_id) 1221 1221 + |> Enum.reverse() 1222 1222 + end 1223 1223 + 1224 1224 + defp parse_generation_created(%DateTime{} = dt), do: {:ok, DateTime.truncate(dt, :second)} 1225 1225 + 1226 1226 + defp parse_generation_created(str) when is_binary(str) do 1227 1227 + case DateTime.from_iso8601(str) do 1228 1228 + {:ok, dt, _offset} -> {:ok, DateTime.truncate(dt, :second)} 1229 1229 + _ -> :error 1230 1230 + end 1231 1231 + end 1232 1232 + 1233 1233 + defp parse_generation_created(_), do: :error 1234 1234 + 1235 1235 + defp normalize_current_generation_rows(rows) do 1236 1236 + current_seed_id = 1237 1237 + Enum.reduce(rows, nil, fn row, acc -> if row.is_current, do: row.seed_id, else: acc end) 1162 1238 1163 1163 - :ok 1164 1164 - end) 1239 1239 + if is_nil(current_seed_id) do 1240 1240 + rows 1241 1241 + else 1242 1242 + Enum.map(rows, fn row -> 1243 1243 + %{row | is_current: row.seed_id == current_seed_id} 1244 1244 + end) 1245 1245 + end 1165 1246 end 1166 1247 1167 1167 - defp upsert_agent_seed_generation(%Agent{} = agent, nix_profile, seed, gen) do 1168 1168 - # If this generation is_current, clear other is_current flags first 1169 1169 - if gen.is_current do 1248 1248 + defp sync_profile_generation_rows(%Agent{} = agent, nix_profile, rows) do 1249 1249 + if Enum.any?(rows, & &1.is_current) do 1170 1250 from(asg in AgentSeedGeneration, 1171 1251 where: asg.agent_id == ^agent.id and asg.profile_id == ^nix_profile.id 1172 1252 ) 1173 1253 |> Repo.update_all(set: [is_current: false]) 1174 1254 end 1175 1255 1176 1176 - # Parse created timestamp 1177 1177 - created_at = 1178 1178 - case gen.created do 1179 1179 - %DateTime{} = dt -> dt 1180 1180 - str when is_binary(str) -> DateTime.from_iso8601(str) |> elem(1) 1181 1181 - end 1256 1256 + keep_seed_ids = 1257 1257 + Enum.reduce(rows, [], fn row, acc -> 1258 1258 + upsert_agent_generation(agent.id, nix_profile.id, row.seed_id, row) 1259 1259 + [row.seed_id | acc] 1260 1260 + end) 1261 1261 + |> Enum.uniq() 1182 1262 1183 1183 - upsert_agent_generation(agent.id, nix_profile.id, seed.id, %{ 1184 1184 - generation_number: gen.generation_number, 1185 1185 - is_current: gen.is_current, 1186 1186 - created_at_generation: created_at 1187 1187 - }) 1263 1263 + delete_stale_agent_seed_generations(agent.id, nix_profile.id, keep_seed_ids) 1264 1264 + end 1265 1265 + 1266 1266 + defp generation_row_changed?(existing, attrs) do 1267 1267 + existing.profile_id != attrs.profile_id or 1268 1268 + existing.generation_number != attrs.generation_number or 1269 1269 + existing.is_current != attrs.is_current or 1270 1270 + existing.created_at_generation != attrs.created_at_generation 1188 1271 end 1189 1272 1190 1273 defp delete_stale_agent_seed_generations(agent_id, profile_id, keep_seed_ids) do

+48

apps/sower/test/sower/orchestration_test.exs

reviewed

··· 807 807 assert hd(profiles).generation_number == 2 808 808 end 809 809 810 810 + test "repeated identical reports do not advance generation id sequence" do 811 811 + agent = agent_fixture() 812 812 + artifact = "/nix/store/#{unique_hash()}-nixos-system-testhost-25.11" 813 813 + created_at = DateTime.to_iso8601(DateTime.utc_now()) 814 814 + 815 815 + report = %SowerClient.Orchestration.AgentSeedsReport{ 816 816 + profiles: [ 817 817 + %SowerClient.Orchestration.AgentSeedProfile{ 818 818 + profile_path: "/nix/var/nix/profiles/system", 819 819 + tags: %{}, 820 820 + generations: [ 821 821 + %SowerClient.Orchestration.AgentSeedGeneration{ 822 822 + path: artifact, 823 823 + link: "/nix/var/nix/profiles/system-42-link", 824 824 + created: created_at, 825 825 + generation_number: 42, 826 826 + is_current: true 827 827 + } 828 828 + ] 829 829 + } 830 830 + ] 831 831 + } 832 832 + 833 833 + assert {:ok, :ok} = Orchestration.update_agent_seed_generations(report, agent) 834 834 + first_sequence_value = agent_seed_generation_sequence_last_value() 835 835 + 836 836 + [first_row] = Orchestration.list_agent_seed_generation(agent) 837 837 + 838 838 + assert {:ok, :ok} = Orchestration.update_agent_seed_generations(report, agent) 839 839 + second_sequence_value = agent_seed_generation_sequence_last_value() 840 840 + 841 841 + [second_row] = Orchestration.list_agent_seed_generation(agent) 842 842 + 843 843 + assert second_row.id == first_row.id 844 844 + assert second_sequence_value == first_sequence_value 845 845 + end 846 846 + 810 847 test "handles multiple profiles (NixOS + home-manager)" do 811 848 agent = agent_fixture() 812 849 nixos_artifact = "/nix/store/#{unique_hash()}-nixos-system-testhost" ··· 866 903 867 904 defp unique_hash do 868 905 :crypto.strong_rand_bytes(16) |> Base.encode32(case: :lower) |> String.slice(0, 32) 906 906 + end 907 907 + 908 908 + defp agent_seed_generation_sequence_last_value do 909 909 + %Postgrex.Result{rows: [[last_value]]} = 910 910 + Ecto.Adapters.SQL.query!( 911 911 + Sower.Repo, 912 912 + "SELECT last_value FROM public.agent_seed_generations_id_seq", 913 913 + [] 914 914 + ) 915 915 + 916 916 + last_value 869 917 end 870 918 end