feat(frontend): mark the public version, open what a listing advertises, and turn pinning on

amber/src/main/scala/org/apache/texera/web/resource/dashboard/user/workflow/WorkflowPublishService.scala

@@ -22,14 +22,16 @@ package org.apache.texera.web.resource.dashboard.user.workflow
 import com.typesafe.scalalogging.LazyLogging
 import org.apache.texera.amber.util.JSONUtils.objectMapper
 import org.apache.texera.dao.SqlServer
-import org.apache.texera.dao.jooq.generated.Tables.WORKFLOW
+import org.apache.texera.dao.jooq.generated.Tables.{WORKFLOW, WORKFLOW_VERSION}
 import org.apache.texera.dao.jooq.generated.enums.DefaultViewEnum
 import org.apache.texera.dao.jooq.generated.tables.daos.WorkflowDao
 import org.apache.texera.dao.jooq.generated.tables.pojos.Workflow
 import org.jooq.{Condition, DSLContext}
 
 import javax.ws.rs.{BadRequestException, NotFoundException}
 import scala.jdk.CollectionConverters.CollectionHasAsScala
+
+import java.sql.Timestamp
 import scala.util.Try
 
 /**
@@ -54,10 +56,14 @@ object WorkflowPublishService extends LazyLogging {
     * What the share dialog asks about: whether the workflow is public, whether a version is pinned,
     * and whether that pin is holding edits back -- the last is true when pinning again would publish
     * something, and always false while following.
+    *
+    * @param pinnedVersionTime the date naming the pinned version in both the dialog and the
+    *                           revision panel.
     */
   case class PublishStatus(
       isPublished: Boolean,
       isPinned: Boolean,
+      pinnedVersionTime: Option[Timestamp],
       hasUnpublishedChanges: Boolean
   )
 
@@ -112,11 +118,14 @@ object WorkflowPublishService extends LazyLogging {
     * two statements, so a `/private` landing between them would otherwise be undone here, putting a
     * workflow the author had just taken down back on show under a frozen copy.
     *
-    * @return how many rows it matched -- zero for a workflow that is missing or not published.
+    * @return how many rows it matched -- zero for a workflow that is missing or not published.    *
+    * `versionId` names the row in the revision history that replays to this copy, so the history can
+    * mark which version is the published one.
     */
-  private def writePin(wid: Integer): Int =
-    context
+  private def writePin(ctx: DSLContext, wid: Integer, versionId: Integer): Unit =
+    ctx
       .update(WORKFLOW)
+      .set(WORKFLOW.PUBLISHED_VERSION_ID, versionId)
       .set(WORKFLOW.PUBLISHED_CONTENT, WORKFLOW.CONTENT)
       .set(WORKFLOW.PUBLISHED_NAME, WORKFLOW.NAME)
       .set(WORKFLOW.PUBLISHED_DESCRIPTION, WORKFLOW.DESCRIPTION)
@@ -129,9 +138,6 @@ object WorkflowPublishService extends LazyLogging {
     * row only with every frozen column set on a public workflow, or with every one of them NULL, so
     * clearing them one at a time -- or clearing them after `is_public` -- would be rejected.
     *
-    * `published_version_id` is named by that constraint as well but is not touched here, for the
-    * same reason [[writePin]] does not set it: nothing writes it yet, so it is NULL on every row.
-    *
     * Dropping a pin is offered on a published workflow only, so that condition travels in the same
     * statement: a `/private` landing between the guard and here cannot be answered with "unpinned,
     * following latest" about a workflow that is now down. Unpublishing carries no such condition --
@@ -142,6 +148,7 @@ object WorkflowPublishService extends LazyLogging {
   private def clearPin(wid: Integer, alsoUnpublish: Boolean = false): Int = {
     val cleared = context
       .update(WORKFLOW)
+      .set(WORKFLOW.PUBLISHED_VERSION_ID, null.asInstanceOf[Integer])
       .set(WORKFLOW.PUBLISHED_CONTENT, null.asInstanceOf[String])
       .set(WORKFLOW.PUBLISHED_NAME, null.asInstanceOf[String])
       .set(WORKFLOW.PUBLISHED_DESCRIPTION, null.asInstanceOf[String])
@@ -161,9 +168,35 @@ object WorkflowPublishService extends LazyLogging {
     * matching means one of two things, and the caller is told which.
     */
   def pinLatest(wid: Integer): PublishStatus = {
-    if (writePin(wid) == 0) {
-      requireWorkflow(wid) // 404s if it is gone
-      throw new BadRequestException(s"Workflow $wid is not published")
+    context.transaction { txConfig =>
+      val ctx = org.jooq.impl.DSL.using(txConfig)
+      // Locked, not merely read: two pins racing would both read before either wrote, and both
+      // insert an anchor. The lock is also what holds the publication check below: a `/private`
+      // arriving after it has to wait for this transaction rather than slipping under it.
+      val workflow = ctx
+        .selectFrom(WORKFLOW)
+        .where(WORKFLOW.WID.eq(wid))
+        .forUpdate()
+        .fetchOneInto(classOf[Workflow])
+      if (workflow == null) {
+        throw new NotFoundException(s"Workflow $wid not found")
+      }
+      // Before the anchor, not after: a pin that cannot land must not leave a version row behind.
+      if (!workflow.getIsPublic) {
+        throw new BadRequestException(s"Workflow $wid is not published")
+      }
+
+      // An anchor in the revision history for the copy being pinned: its delta is the identity patch,
+      // so replaying this row returns what was published however many edits pile up later. An
+      // existing version row cannot stand in, since one replays to the content as it was *before*
+      // the change it records. Pinning again unchanged reuses the anchor rather than adding a twin.
+      val anchorVid = Option(workflow.getPublishedVersionId)
+        .filter(_ =>
+          Option(workflow.getPublishedContent).exists(sameContent(_, workflow.getContent))
+        )
+        .getOrElse(WorkflowVersionResource.insertNewVersion(wid, ctx = ctx).getVid)
+
+      writePin(ctx, wid, anchorVid)
     }
     logger.info(s"Workflow $wid pinned to its latest content")
     statusOf(wid)
@@ -193,13 +226,30 @@ object WorkflowPublishService extends LazyLogging {
     logger.info(s"Workflow $wid unpublished")
   }
 
+  /** Read from the version row, so the dialog and the revision panel print one date rather than two. */
+  private def pinnedVersionTimeOf(versionId: Integer): Option[Timestamp] =
+    Option(
+      context
+        .select(WORKFLOW_VERSION.CREATION_TIME)
+        .from(WORKFLOW_VERSION)
+        .where(WORKFLOW_VERSION.VID.eq(versionId))
+        .fetchOneInto(classOf[Timestamp])
+    )
+
+  /** The date a public viewer should see: that of the version on show, not of an edit they cannot. */
+  def publicModifiedTime(workflow: Workflow): Timestamp =
+    Option(workflow.getPublishedVersionId)
+      .flatMap(pinnedVersionTimeOf)
+      .getOrElse(workflow.getLastModifiedTime)
+
   /** Whether a version is pinned, and whether it is holding edits back. */
   def statusOf(wid: Integer): PublishStatus = {
     val workflow = requireWorkflow(wid)
     val pinned = workflow.getPublishedContent != null
     PublishStatus(
       isPublished = workflow.getIsPublic,
       isPinned = pinned,
+      pinnedVersionTime = Option(workflow.getPublishedVersionId).flatMap(pinnedVersionTimeOf),
       // Literally "what the public sees is not what you have": whatever [[publicCopyOf]] freezes is
       // what this compares, on values rather than version ids, so an edit and its undo cancel out.
       hasUnpublishedChanges = differs(publicCopyOf(workflow), workingCopyOf(workflow))

amber/src/main/scala/org/apache/texera/web/resource/dashboard/user/workflow/WorkflowResource.scala

@@ -720,6 +720,10 @@ class WorkflowResource extends LazyLogging {
     )
   }
 
+  /**
+    * Publishes the workflow. The public follows the author's latest content until a version is
+    * pinned; see [[pinLatest]].
+    */
   @PUT
   @RolesAllowed(Array("REGULAR", "ADMIN"))
   @Path("/public/{wid}")
@@ -956,14 +960,16 @@ class WorkflowResource extends LazyLogging {
     }
     // Name and description come from the public copy for the same reason as the content: a pin has
     // to hold everything on show.
-    val publicCopy = WorkflowPublishService.publicCopyOf(workflow.into(classOf[Workflow]))
+    val stored = workflow.into(classOf[Workflow])
+    val publicCopy = WorkflowPublishService.publicCopyOf(stored)
     WorkflowWithPrivilege(
       publicCopy.name,
       publicCopy.description,
       workflow.getWid,
       publicCopy.content,
       workflow.getCreationTime,
-      workflow.getLastModifiedTime,
+      // Dated by the version on show, not by the author's most recent private edit.
+      WorkflowPublishService.publicModifiedTime(stored),
       workflow.getIsPublic,
       readonly = true,
       // The view freezes with the copy: the form's definition lives inside the content.

amber/src/main/scala/org/apache/texera/web/resource/dashboard/user/workflow/WorkflowVersionResource.scala

@@ -127,11 +127,15 @@ object WorkflowVersionResource {
     *
     * @param wid
     */
-  def insertNewVersion(wid: Integer, content: String = "[]"): WorkflowVersion = {
+  def insertNewVersion(
+      wid: Integer,
+      content: String = "[]",
+      ctx: DSLContext = context
+  ): WorkflowVersion = {
     val workflowVersion = new WorkflowVersion()
     workflowVersion.setContent(content)
     workflowVersion.setWid(wid)
-    workflowVersionDao.insert(workflowVersion)
+    new WorkflowVersionDao(ctx.configuration).insert(workflowVersion)
     workflowVersion
   }
 
@@ -195,39 +199,45 @@ object WorkflowVersionResource {
     * @return
     */
   private def encodeVersionImportance(
-      currentVersions: List[WorkflowVersion]
+      currentVersions: List[WorkflowVersion],
+      publicVersionId: Option[Integer]
   ): List[VersionEntry] = {
     var impEncodedVersions: List[VersionEntry] = List()
 
+    // A pin's anchor is not a version the author made: it carries the identity patch and appears the
+    // moment they pin. It is always shown, so they can find and restore what the public has, but it
+    // must not start the aggregation window -- the save it was pinned from is seconds older, and
+    // would otherwise be folded into it and disappear from the panel.
+    def isAnchor(version: WorkflowVersion): Boolean = publicVersionId.contains(version.getVid)
+
     val lastVersion = currentVersions.head
-    var lastVersionTime = lastVersion.getCreationTime
+    var lastVersionTime: Option[Timestamp] =
+      if (isAnchor(lastVersion)) None else Some(lastVersion.getCreationTime)
     impEncodedVersions = impEncodedVersions :+ VersionEntry(
       lastVersion.getVid,
       lastVersion.getCreationTime,
       lastVersion.getContent,
-      true
-    ) // the first (latest)
-    // version is important even if it is positional
+      true,
+      isAnchor(lastVersion)
+    ) // the first (latest) version is important even if it is positional
     var versionImportance: Boolean = true
     for (version <- currentVersions.tail) {
-      if (
-        isWithinTimeLimit(
-          lastVersionTime,
-          version.getCreationTime
-        )
-      ) {
+      if (isAnchor(version)) {
+        versionImportance = true
+      } else if (lastVersionTime.exists(isWithinTimeLimit(_, version.getCreationTime))) {
         versionImportance = false
       } // try reducing unnecessary check of positional versions
       // because parsing the Json string is expensive
       else {
-        lastVersionTime = version.getCreationTime
+        lastVersionTime = Some(version.getCreationTime)
         versionImportance = isVersionImportant(version.getContent)
       }
       impEncodedVersions = impEncodedVersions :+ VersionEntry(
         version.getVid,
         version.getCreationTime,
         version.getContent,
-        versionImportance
+        versionImportance,
+        isAnchor(version)
       )
     }
     impEncodedVersions
@@ -326,7 +336,10 @@ object WorkflowVersionResource {
       vId: Integer,
       creationTime: Timestamp,
       content: String,
-      importance: Boolean
+      importance: Boolean,
+      // True for the one version the Hub is serving right now, so the author can tell at a glance
+      // where the public copy sits relative to what they are editing.
+      isCurrentlyPublic: Boolean
   )
 
 }
@@ -360,7 +373,8 @@ class WorkflowVersionResource {
           .orderBy(WORKFLOW_VERSION.CREATION_TIME.desc())
           .fetchInto(classOf[WorkflowVersion])
           .asScala
-          .toList
+          .toList,
+        Option(workflowDao.fetchOneByWid(wid)).flatMap(w => Option(w.getPublishedVersionId))
       )
     }
   }