feat(workflow): match search and listings against the copy on show

amber/src/main/scala/org/apache/texera/web/resource/dashboard/SearchQueryBuilder.scala

@@ -58,7 +58,17 @@ trait SearchQueryBuilder {
       includePublic: Boolean = false
   ): TableLike[_]
 
-  protected def constructWhereClause(uid: Integer, params: SearchQueryParams): Condition
+  /**
+    * @param includePublic whether public resources the user has not been granted access to are in
+    *                      scope. Builders whose content differs between the public and the private
+    *                      view (workflows, whose public copy is a pinned version) need this to know
+    *                      which copy a filter may match against.
+    */
+  protected def constructWhereClause(
+      uid: Integer,
+      params: SearchQueryParams,
+      includePublic: Boolean = false
+  ): Condition
 
   protected def getGroupByFields: Seq[GroupField] = Seq.empty
 
@@ -78,7 +88,7 @@ trait SearchQueryBuilder {
     val query: SelectGroupByStep[Record] = context
       .selectDistinct(mappedResourceSchema.allFields: _*)
       .from(constructFromClause(uid, params, includePublic))
-      .where(constructWhereClause(uid, params))
+      .where(constructWhereClause(uid, params, includePublic))
     val groupByFields = getGroupByFields
     if (groupByFields.nonEmpty) {
       query.groupBy(groupByFields: _*)

amber/src/main/scala/org/apache/texera/web/resource/dashboard/UnifiedResourceSchema.scala

@@ -80,7 +80,12 @@ object UnifiedResourceSchema {
       // mark the row and route it accordingly.
       workflowDefaultView: Field[DefaultViewEnum] = DSL.cast(null, classOf[DefaultViewEnum]),
       modelFramework: Field[String] = DSL.cast(null, classOf[String]),
-      modelFormat: Field[String] = DSL.cast(null, classOf[String])
+      modelFormat: Field[String] = DSL.cast(null, classOf[String]),
+      workflowHasUnpublishedChanges: Field[java.lang.Boolean] =
+        DSL.cast(null, classOf[java.lang.Boolean]),
+      workflowPublishedName: Field[String] = DSL.cast(null, classOf[String]),
+      workflowPublishedDescription: Field[String] = DSL.cast(null, classOf[String]),
+      viewerHasGrantedAccess: Field[java.lang.Boolean] = DSL.cast(null, classOf[java.lang.Boolean])
   ): UnifiedResourceSchema = {
     new UnifiedResourceSchema(
       Seq(
@@ -110,7 +115,13 @@ object UnifiedResourceSchema {
         workflowCoverImage -> workflowCoverImage.as("workflow_cover_image"),
         workflowDefaultView -> workflowDefaultView.as("workflow_default_view"),
         modelFramework -> modelFramework.as("model_framework"),
-        modelFormat -> modelFormat.as("model_format")
+        modelFormat -> modelFormat.as("model_format"),
+        workflowHasUnpublishedChanges -> workflowHasUnpublishedChanges
+          .as("workflow_has_unpublished_changes"),
+        workflowPublishedName -> workflowPublishedName.as("workflow_published_name"),
+        workflowPublishedDescription -> workflowPublishedDescription
+          .as("workflow_published_description"),
+        viewerHasGrantedAccess -> viewerHasGrantedAccess.as("viewer_has_granted_access")
       )
     )
   }

amber/src/main/scala/org/apache/texera/web/resource/dashboard/VersionedResourceSearchQueryBuilder.scala

@@ -110,7 +110,8 @@ abstract class VersionedResourceSearchQueryBuilder[Rec <: Record, P](
 
   override protected def constructWhereClause(
       uid: Integer,
-      params: DashboardResource.SearchQueryParams
+      params: DashboardResource.SearchQueryParams,
+      includePublic: Boolean
   ): Condition = {
     val splitKeywords = params.keywords.asScala
       .flatMap(_.split("[+\\-()<>~*@\"]"))

amber/src/main/scala/org/apache/texera/web/resource/dashboard/WorkflowSearchQueryBuilder.scala

@@ -23,9 +23,10 @@ import org.apache.texera.dao.jooq.generated.Tables._
 import org.apache.texera.dao.jooq.generated.tables.pojos.Workflow
 import org.apache.texera.web.resource.dashboard.DashboardResource.DashboardClickableFileEntry
 import org.apache.texera.web.resource.dashboard.FulltextSearchQueryUtils._
+import org.apache.texera.web.resource.dashboard.user.workflow.WorkflowPublishService
 import org.apache.texera.web.resource.dashboard.user.workflow.WorkflowResource.DashboardWorkflow
 import org.jooq.impl.DSL
-import org.jooq.{Condition, GroupField, Record, TableLike}
+import org.jooq.{Condition, Field, GroupField, Record, TableLike}
 
 import scala.jdk.CollectionConverters.CollectionHasAsScala
 import org.apache.texera.dao.jooq.generated.enums.{DefaultViewEnum, PrivilegeEnum}
@@ -53,7 +54,25 @@ object WorkflowSearchQueryBuilder extends SearchQueryBuilder {
       ownerId = WORKFLOW_OF_USER.UID,
       userName = USER.NAME,
       workflowCoverImage = DSL.max(WORKFLOW_COVER_IMAGE.IMAGE).as("workflow_cover_image"),
-      workflowDefaultView = WORKFLOW.DEFAULT_VIEW.as("workflow_default_view")
+      workflowDefaultView = WORKFLOW.DEFAULT_VIEW.as("workflow_default_view"),
+      // The isNotNull guard because isDistinctFrom would read a NULL pin as different content, and
+      // every unpinned public workflow would report drift. Aggregated to stay out of the GROUP BY,
+      // which would otherwise group the search by two TEXT columns.
+      workflowHasUnpublishedChanges = DSL
+        .boolOr(
+          WORKFLOW.PUBLISHED_CONTENT.isNotNull
+            .and(WorkflowPublishService.pinDiffersFromWorkingCopy)
+        )
+        .as("workflow_has_unpublished_changes"),
+      // What a viewer without granted access is shown instead of the author's live metadata. NULL
+      // while following, which the reader treats as "show the live values".
+      workflowPublishedName = DSL.max(WORKFLOW.PUBLISHED_NAME).as("workflow_published_name"),
+      workflowPublishedDescription =
+        DSL.max(WORKFLOW.PUBLISHED_DESCRIPTION).as("workflow_published_description"),
+      // The access join is already restricted to the caller, so this needs no user id of its own.
+      viewerHasGrantedAccess = DSL
+        .boolOr(WORKFLOW_USER_ACCESS.UID.isNotNull)
+        .as("viewer_has_granted_access")
     )
   }
 
@@ -88,9 +107,58 @@ object WorkflowSearchQueryBuilder extends SearchQueryBuilder {
     baseQuery.where(condition)
   }
 
+  /** Rows the user was granted access to, as opposed to rows they see only because it is public. */
+  private def grantedAccessCondition(uid: Integer): Condition =
+    if (uid == null) DSL.falseCondition()
+    else WORKFLOW_USER_ACCESS.UID.eq(uid)
+
+  /**
+    * The three searchable columns carrying one copy. They travel together because a pin freezes them
+    * together: a filter over content alone would match a pinned workflow on a title no public viewer
+    * has seen.
+    */
+  private case class WorkflowCopy(
+      name: Field[String],
+      description: Field[String],
+      content: Field[String]
+  )
+
+  private val workingCopy =
+    WorkflowCopy(WORKFLOW.NAME, WORKFLOW.DESCRIPTION, WORKFLOW.CONTENT)
+  private val pinnedCopy =
+    WorkflowCopy(
+      WORKFLOW.PUBLISHED_NAME,
+      WORKFLOW.PUBLISHED_DESCRIPTION,
+      WORKFLOW.PUBLISHED_CONTENT
+    )
+
+  /**
+    * Applies a filter to whichever copy the user is allowed to see: one search can return both their
+    * own workflows and public ones, and a pinned public one must not turn up on keywords that exist
+    * only behind the pin. A disjunction of guarded filters over bare columns rather than a CASE, so
+    * each side stays eligible for its own PGroonga index.
+    */
+  private def onVisibleCopy(
+      uid: Integer,
+      includePublic: Boolean
+  )(build: WorkflowCopy => Condition): Condition = {
+    val onWorkingCopy = build(workingCopy).and(grantedAccessCondition(uid))
+    val onPublicCopy = WORKFLOW.IS_PUBLIC
+      .eq(true)
+      .and(
+        // Following leaves the pinned columns NULL, and the public copy is then the working one.
+        build(pinnedCopy)
+          .or(WORKFLOW.PUBLISHED_CONTENT.isNull.and(build(workingCopy)))
+      )
+    if (uid == null) onPublicCopy
+    else if (includePublic) onWorkingCopy.or(onPublicCopy)
+    else onWorkingCopy
+  }
+
   override protected def constructWhereClause(
       uid: Integer,
-      params: DashboardResource.SearchQueryParams
+      params: DashboardResource.SearchQueryParams,
+      includePublic: Boolean
   ): Condition = {
     val splitKeywords = params.keywords.asScala
       .flatMap(_.split("[+\\-()<>~*@\"]"))
@@ -114,13 +182,23 @@ object WorkflowSearchQueryBuilder extends SearchQueryBuilder {
       // Apply owner filter
       .and(getContainsFilter(params.owners, USER.EMAIL))
       // Apply operators filter
-      .and(getOperatorsFilter(params.operators, WORKFLOW.CONTENT))
+      .and(
+        if (params.operators.isEmpty) DSL.noCondition()
+        else
+          onVisibleCopy(uid, includePublic)(copy =>
+            getOperatorsFilter(params.operators, copy.content)
+          )
+      )
       // Apply fulltext search filter
       .and(
-        getFullTextSearchFilter(
-          splitKeywords,
-          List(WORKFLOW.NAME, WORKFLOW.DESCRIPTION, WORKFLOW.CONTENT)
-        )
+        if (splitKeywords.isEmpty) DSL.noCondition()
+        else
+          onVisibleCopy(uid, includePublic)(copy =>
+            getFullTextSearchFilter(
+              splitKeywords,
+              List(copy.name, copy.description, copy.content)
+            )
+          )
       )
   }
 
@@ -141,20 +219,35 @@ object WorkflowSearchQueryBuilder extends SearchQueryBuilder {
       uid: Integer,
       record: Record
   ): DashboardResource.DashboardClickableFileEntry = {
+    val workflow = record.into(WORKFLOW).into(classOf[Workflow])
+    // The select lists specific columns, so the POJO built from the record does not carry this one.
+    // Without it the listing forgets the default-view preference on every refresh.
+    workflow.setDefaultView(record.get("workflow_default_view", classOf[DefaultViewEnum]))
+
+    // A viewer here only because the workflow is public sees the pinned name and description, the
+    // same copy the detail page serves -- otherwise a listing would advertise a title that opening it
+    // does not show. Both are NULL while following, which leaves the live values in place.
+    // Unknown counts as not granted: the reverse is the leak.
+    val granted = Option(record.get("viewer_has_granted_access", classOf[java.lang.Boolean]))
+      .exists(_.booleanValue())
+    if (!granted) {
+      Option(record.get("workflow_published_name", classOf[String])).foreach(workflow.setName)
+      Option(record.get("workflow_published_description", classOf[String]))
+        .foreach(workflow.setDescription)
+    }
+
     val dw = DashboardWorkflow(
       record.into(WORKFLOW_OF_USER).getUid == uid,
       Option(record.get(WORKFLOW_USER_ACCESS.PRIVILEGE, classOf[PrivilegeEnum]))
         .map(_.toString)
         .getOrElse(PrivilegeEnum.NONE.toString),
-      record.into(USER).getName, {
-        // The select lists specific columns, so the POJO built from the record does not carry
-        // this one. Without it the listing forgets the default-view preference on every refresh.
-        val w = record.into(WORKFLOW).into(classOf[Workflow])
-        w.setDefaultView(record.get("workflow_default_view", classOf[DefaultViewEnum]))
-        w
-      },
+      record.into(USER).getName,
+      workflow,
       record.into(USER).getUid,
-      Option(record.get("workflow_cover_image", classOf[String]))
+      Option(record.get("workflow_cover_image", classOf[String])),
+      // Null for the resource types that do not define the column at all.
+      Option(record.get("workflow_has_unpublished_changes", classOf[java.lang.Boolean]))
+        .exists(_.booleanValue())
     )
     DashboardClickableFileEntry(SearchQueryBuilder.WORKFLOW_RESOURCE_TYPE, workflow = Some(dw))
   }

amber/src/main/scala/org/apache/texera/web/resource/dashboard/hub/HubResource.scala

@@ -30,6 +30,7 @@ import org.apache.texera.web.resource.dashboard.VersionedResourceTables
 import org.apache.texera.web.resource.dashboard.hub.ActionType.{Clone, Like, Unlike, View}
 import org.apache.texera.web.resource.dashboard.hub.EntityTables._
 import org.apache.texera.web.resource.dashboard.hub.HubResource._
+import org.apache.texera.web.resource.dashboard.user.workflow.WorkflowPublishService
 import org.apache.texera.web.resource.dashboard.user.workflow.WorkflowResource.{
   DashboardWorkflow,
   baseWorkflowSelect,
@@ -273,7 +274,23 @@ object HubResource {
       )
       .fetch()
 
-    mapWorkflowEntries(records, uid)
+    // The hub is the public shelf, so everything on it is listed as the public sees it: while a pin
+    // is in place, under the name and description frozen with it rather than the author's live ones
+    // -- for the author too, who is looking at the shelf and not at their own dashboard.
+    val entries = mapWorkflowEntries(records, uid)
+    val pinned = WorkflowPublishService.pinnedListingsOf(entries.map(_.workflow.getWid))
+    entries.map { entry =>
+      pinned.get(entry.workflow.getWid) match {
+        case None => entry
+        case Some(listing) =>
+          entry.workflow.setName(listing.name)
+          entry.workflow.setDescription(listing.description)
+          // Carried like the search listing carries it: a card advertising the pinned copy has to
+          // open that copy, and without this flag the author's own card would open their editor and
+          // show them something else.
+          entry.copy(hasUnpublishedChanges = listing.hasUnpublishedChanges)
+      }
+    }
   }
 
   /**

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

@@ -92,6 +92,15 @@ object WorkflowAccessResource {
     }
   }
 
+  /**
+    * Whether the user was granted access to the workflow itself, rather than merely being able to
+    * read it because it is public. Granted access sees the author's working copy; public access is
+    * held at the pinned copy while one is pinned.
+    */
+  def hasGrantedAccess(wid: Integer, uid: Integer): Boolean = {
+    !getPrivilege(wid, uid).eq(PrivilegeEnum.NONE)
+  }
+
   def isPublic(wid: Integer): Boolean = {
     context
       .select(WORKFLOW.IS_PUBLIC)