-
Notifications
You must be signed in to change notification settings - Fork 74
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Add the project last update projection for incoming passivation featu…
…re (#5172) * Add the project last update projection for incoming passivation feature --------- Co-authored-by: Simon Dumas <simon.dumas@epfl.ch>
- Loading branch information
Showing
19 changed files
with
464 additions
and
50 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
16 changes: 16 additions & 0 deletions
16
delta/kernel/src/main/scala/ch/epfl/bluebrain/nexus/delta/kernel/utils/CollectionUtils.scala
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,16 @@ | ||
| package ch.epfl.bluebrain.nexus.delta.kernel.utils | ||
|
|
||
| object CollectionUtils { | ||
|
|
||
| /** | ||
| * Displays all elements of this collection between quotes and separated by commas. | ||
| * @param iterable | ||
| * the collection to display | ||
| * @return | ||
| * a string representation of the colleciton | ||
| * @example | ||
| * `CollectionUtils.quote(List(1, 2, 3)) = "'1','2','3'"` | ||
| */ | ||
| def quote(iterable: Iterable[_]): String = iterable.mkString("'", "','", "'") | ||
|
|
||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
10 changes: 10 additions & 0 deletions
10
...ing-psql/src/main/resources/scripts/postgres/init/V1_11_M03_001__project_last_updates.ddl
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,10 @@ | ||
| CREATE TABLE IF NOT EXISTS public.project_last_updates( | ||
| org text NOT NULL, | ||
| project text NOT NULL, | ||
| last_instant timestamptz NOT NULL, | ||
| last_state_ordering bigint NOT NULL, | ||
| PRIMARY KEY(org, project) | ||
| ); | ||
|
|
||
| CREATE INDEX IF NOT EXISTS project_last_updates_last_instant_idx ON public.project_last_updates(last_instant); | ||
|
|
11 changes: 11 additions & 0 deletions
11
...rc/main/scala/ch/epfl/bluebrain/nexus/delta/sourcing/config/ProjectLastUpdateConfig.scala
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,11 @@ | ||
| package ch.epfl.bluebrain.nexus.delta.sourcing.config | ||
|
|
||
| import pureconfig.ConfigReader | ||
| import pureconfig.generic.semiauto.deriveReader | ||
|
|
||
| final case class ProjectLastUpdateConfig(batch: BatchConfig, query: QueryConfig) | ||
|
|
||
| object ProjectLastUpdateConfig { | ||
| implicit final val projectLastUpdateConfig: ConfigReader[ProjectLastUpdateConfig] = | ||
| deriveReader[ProjectLastUpdateConfig] | ||
| } |
80 changes: 80 additions & 0 deletions
80
...ain/scala/ch/epfl/bluebrain/nexus/delta/sourcing/projections/ProjectLastUpdateStore.scala
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,80 @@ | ||
| package ch.epfl.bluebrain.nexus.delta.sourcing.projections | ||
|
|
||
| import cats.effect.IO | ||
| import cats.syntax.all._ | ||
| import ch.epfl.bluebrain.nexus.delta.sourcing.Transactors | ||
| import ch.epfl.bluebrain.nexus.delta.sourcing.model.ProjectRef | ||
| import ch.epfl.bluebrain.nexus.delta.sourcing.projections.model.ProjectLastUpdate | ||
| import ch.epfl.bluebrain.nexus.delta.sourcing.projections.model.ProjectLastUpdate.ProjectLastUpdateMap | ||
| import doobie.Fragments | ||
| import doobie.syntax.all._ | ||
| import doobie.postgres.implicits._ | ||
|
|
||
| import java.time.Instant | ||
|
|
||
| /** | ||
| * Keeps track of the last update on a given project | ||
| */ | ||
| trait ProjectLastUpdateStore { | ||
|
|
||
| /** | ||
| * * Delete the entry for the given project | ||
| */ | ||
| def delete(project: ProjectRef): IO[Unit] | ||
|
|
||
| /** | ||
| * Inserts/updates a list of updates | ||
| */ | ||
| def save(updates: List[ProjectLastUpdate]): IO[Unit] | ||
|
|
||
| /** | ||
| * Fetch all updates from the database | ||
| */ | ||
| def fetchAll: IO[ProjectLastUpdateMap] | ||
|
|
||
| /** | ||
| * Fetch updates older than the given instant | ||
| */ | ||
| def fetchUpdates(after: Instant): IO[ProjectLastUpdateMap] | ||
|
|
||
| } | ||
|
|
||
| object ProjectLastUpdateStore { | ||
|
|
||
| def apply(xas: Transactors): ProjectLastUpdateStore = new ProjectLastUpdateStore { | ||
|
|
||
| override def delete(project: ProjectRef): IO[Unit] = | ||
| sql"""DELETE FROM project_last_updates WHERE org = ${project.organization} and project = ${project.project}""".update.run | ||
| .transact(xas.write) | ||
| .void | ||
|
|
||
| override def save(updates: List[ProjectLastUpdate]): IO[Unit] = | ||
| updates | ||
| .traverse(saveOne) | ||
| .transact(xas.write) | ||
| .void | ||
|
|
||
| private def saveOne(p: ProjectLastUpdate) = | ||
| sql"""INSERT INTO project_last_updates (org, project, last_instant, last_state_ordering) | ||
| |VALUES (${p.project.organization}, ${p.project.project} ,${p.lastInstant}, ${p.lastOrdering}) | ||
| |ON CONFLICT (org, project) | ||
| |DO UPDATE set | ||
| | last_instant = EXCLUDED.last_instant, | ||
| | last_state_ordering = EXCLUDED.last_state_ordering; | ||
| |""".stripMargin.update.run | ||
|
|
||
| override def fetchAll: IO[ProjectLastUpdateMap] = fetch(None) | ||
|
|
||
| override def fetchUpdates(after: Instant): IO[ProjectLastUpdateMap] = fetch(Some(after)) | ||
|
|
||
| private def fetch(after: Option[Instant]) = { | ||
| val afterFragment = after.map { a => fr"last_instant > $a" } | ||
| sql"""SELECT * from project_last_updates ${Fragments.whereAndOpt(afterFragment)}""" | ||
| .query[ProjectLastUpdate] | ||
| .map { plu => plu.project -> plu } | ||
| .toMap | ||
| .transact(xas.read) | ||
| } | ||
| } | ||
|
|
||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
21 changes: 21 additions & 0 deletions
21
...in/scala/ch/epfl/bluebrain/nexus/delta/sourcing/projections/model/ProjectLastUpdate.scala
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,21 @@ | ||
| package ch.epfl.bluebrain.nexus.delta.sourcing.projections.model | ||
|
|
||
| import ch.epfl.bluebrain.nexus.delta.sourcing.model.{Label, ProjectRef} | ||
| import ch.epfl.bluebrain.nexus.delta.sourcing.offset.Offset | ||
| import doobie.postgres.implicits._ | ||
| import doobie.Read | ||
|
|
||
| import java.time.Instant | ||
|
|
||
| final case class ProjectLastUpdate(project: ProjectRef, lastInstant: Instant, lastOrdering: Offset) | ||
|
|
||
| object ProjectLastUpdate { | ||
|
|
||
| type ProjectLastUpdateMap = Map[ProjectRef, ProjectLastUpdate] | ||
|
|
||
| implicit val projectLastUpdateRead: Read[ProjectLastUpdate] = | ||
| Read[(Label, Label, Instant, Offset)].map { case (org, project, instant, offset) => | ||
| ProjectLastUpdate(ProjectRef(org, project), instant, offset) | ||
| } | ||
|
|
||
| } |
Oops, something went wrong.