From f44c37f6bae22732c980ecf4bf1bcde45e8b1a7c Mon Sep 17 00:00:00 2001 From: Yang Zhang Date: Sun, 18 Oct 2026 15:06:22 -0700 Subject: [PATCH 1/2] feat(workflow): match search and listings against the copy on show Search and the listings it feeds were reading the author's live columns, which for a pinned workflow is the one copy the public cannot open. A draft would turn up in a public search under a title nobody has seen, and the card would advertise a name the detail page does not show. Each filter is now applied to whichever copy the caller may see: `onVisibleCopy` builds the same filter twice -- over the live columns for rows the caller was granted access to, over the frozen ones for rows they reach only because the workflow is public -- and ORs the two. A disjunction over bare columns rather than a CASE, so each side stays eligible for its own fulltext index. Unpinned public rows fall back to the live columns, so a following workflow searches exactly as it does now. Listings carry two more things from the same query: the frozen name and description to show a viewer without granted access, and whether the copy on show is behind the author's working copy. `constructWhereClause` takes `includePublic` for this; the other builders accept and ignore it. Co-Authored-By: Claude Opus 5 --- .../dashboard/SearchQueryBuilder.scala | 14 +- .../dashboard/UnifiedResourceSchema.scala | 19 +- .../VersionedResourceSearchQueryBuilder.scala | 3 +- .../WorkflowSearchQueryBuilder.scala | 131 +++++++- .../resource/dashboard/hub/HubResource.scala | 22 +- .../workflow/WorkflowPublishService.scala | 61 +++- .../user/workflow/WorkflowResource.scala | 5 +- .../dashboard/UnifiedResourceSchemaSpec.scala | 35 +- .../WorkflowSearchQueryBuilderSpec.scala | 88 ++++- .../user/workflow/WorkflowPublishSpec.scala | 308 +++++++++++++++++- 10 files changed, 647 insertions(+), 39 deletions(-) diff --git a/amber/src/main/scala/org/apache/texera/web/resource/dashboard/SearchQueryBuilder.scala b/amber/src/main/scala/org/apache/texera/web/resource/dashboard/SearchQueryBuilder.scala index c132500625a..76c6e7d9763 100644 --- a/amber/src/main/scala/org/apache/texera/web/resource/dashboard/SearchQueryBuilder.scala +++ b/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: _*) diff --git a/amber/src/main/scala/org/apache/texera/web/resource/dashboard/UnifiedResourceSchema.scala b/amber/src/main/scala/org/apache/texera/web/resource/dashboard/UnifiedResourceSchema.scala index 40171c8f11e..d1144f635fa 100644 --- a/amber/src/main/scala/org/apache/texera/web/resource/dashboard/UnifiedResourceSchema.scala +++ b/amber/src/main/scala/org/apache/texera/web/resource/dashboard/UnifiedResourceSchema.scala @@ -80,7 +80,14 @@ 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]), + workflowPublishedDefaultView: Field[DefaultViewEnum] = + DSL.cast(null, classOf[DefaultViewEnum]), + viewerHasGrantedAccess: Field[java.lang.Boolean] = DSL.cast(null, classOf[java.lang.Boolean]) ): UnifiedResourceSchema = { new UnifiedResourceSchema( Seq( @@ -110,7 +117,15 @@ 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"), + workflowPublishedDefaultView -> workflowPublishedDefaultView + .as("workflow_published_default_view"), + viewerHasGrantedAccess -> viewerHasGrantedAccess.as("viewer_has_granted_access") ) ) } diff --git a/amber/src/main/scala/org/apache/texera/web/resource/dashboard/VersionedResourceSearchQueryBuilder.scala b/amber/src/main/scala/org/apache/texera/web/resource/dashboard/VersionedResourceSearchQueryBuilder.scala index aa60f11c8e4..4b693152a1c 100644 --- a/amber/src/main/scala/org/apache/texera/web/resource/dashboard/VersionedResourceSearchQueryBuilder.scala +++ b/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("[+\\-()<>~*@\"]")) diff --git a/amber/src/main/scala/org/apache/texera/web/resource/dashboard/WorkflowSearchQueryBuilder.scala b/amber/src/main/scala/org/apache/texera/web/resource/dashboard/WorkflowSearchQueryBuilder.scala index 2e08a98d579..ca0e80fb4ea 100644 --- a/amber/src/main/scala/org/apache/texera/web/resource/dashboard/WorkflowSearchQueryBuilder.scala +++ b/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,29 @@ 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 view belongs with the name and the description: a card that opens the frozen copy in + // the author's live view would offer a Form View over a graph that carries no form. + workflowPublishedDefaultView = + DSL.max(WORKFLOW.PUBLISHED_DEFAULT_VIEW).as("workflow_published_default_view"), + // 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 +111,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 +186,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 +223,37 @@ 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) + Option(record.get("workflow_published_default_view", classOf[DefaultViewEnum])) + .foreach(workflow.setDefaultView) + } + 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)) } diff --git a/amber/src/main/scala/org/apache/texera/web/resource/dashboard/hub/HubResource.scala b/amber/src/main/scala/org/apache/texera/web/resource/dashboard/hub/HubResource.scala index 42ae99d9ce3..7b4aa41c2f2 100644 --- a/amber/src/main/scala/org/apache/texera/web/resource/dashboard/hub/HubResource.scala +++ b/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,26 @@ 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) + // The view too: a hub card that opened the frozen copy in the author's live view would + // offer a Form View over a graph carrying no form. + entry.workflow.setDefaultView(listing.defaultView) + // 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) + } + } } /** diff --git a/amber/src/main/scala/org/apache/texera/web/resource/dashboard/user/workflow/WorkflowPublishService.scala b/amber/src/main/scala/org/apache/texera/web/resource/dashboard/user/workflow/WorkflowPublishService.scala index 9d6a208fcd0..62d15fde230 100644 --- a/amber/src/main/scala/org/apache/texera/web/resource/dashboard/user/workflow/WorkflowPublishService.scala +++ b/amber/src/main/scala/org/apache/texera/web/resource/dashboard/user/workflow/WorkflowPublishService.scala @@ -26,9 +26,10 @@ import org.apache.texera.dao.jooq.generated.Tables.WORKFLOW 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.DSLContext +import org.jooq.{Condition, DSLContext} import javax.ws.rs.{BadRequestException, NotFoundException} +import scala.jdk.CollectionConverters.CollectionHasAsScala import scala.util.Try /** @@ -248,6 +249,64 @@ object WorkflowPublishService extends LazyLogging { workflow.getDefaultView ) + /** + * [[differs]] as a condition, for the listings that ask about many workflows at once: the same + * fields, so a card and the share dialog can never disagree about whether edits are held back. + * Only meaningful on a row that is pinned -- while following, every frozen column is NULL and + * `isDistinctFrom` would read that as drift. + */ + val pinDiffersFromWorkingCopy: Condition = + WORKFLOW.PUBLISHED_CONTENT + .isDistinctFrom(WORKFLOW.CONTENT) + .or(WORKFLOW.PUBLISHED_NAME.isDistinctFrom(WORKFLOW.NAME)) + .or(WORKFLOW.PUBLISHED_DESCRIPTION.isDistinctFrom(WORKFLOW.DESCRIPTION)) + .or(WORKFLOW.PUBLISHED_DEFAULT_VIEW.isDistinctFrom(WORKFLOW.DEFAULT_VIEW)) + + /** + * What a listing needs about one pinned workflow: the frozen name and description it must show + * instead of the author's live ones, and whether those live ones have moved on. + */ + case class PinnedListing( + name: String, + description: String, + defaultView: DefaultViewEnum, + hasUnpublishedChanges: Boolean + ) + + /** + * The pinned listings among `wids`, keyed by wid. A workflow that follows the author's latest is + * simply absent, which leaves its live values in place and its drift flag false. + * + * Drift is decided in SQL here rather than by [[differs]], because a listing asks about many + * workflows at once and none of their contents are worth shipping back to compare in memory. + */ + def pinnedListingsOf(wids: Seq[Integer]): Map[Integer, PinnedListing] = + if (wids.isEmpty) Map() + else { + val drifted = pinDiffersFromWorkingCopy + context + .select( + WORKFLOW.WID, + WORKFLOW.PUBLISHED_NAME, + WORKFLOW.PUBLISHED_DESCRIPTION, + WORKFLOW.PUBLISHED_DEFAULT_VIEW, + drifted + ) + .from(WORKFLOW) + .where(WORKFLOW.WID.in(wids: _*).and(WORKFLOW.PUBLISHED_CONTENT.isNotNull)) + .fetch() + .asScala + .map(row => + row.get(WORKFLOW.WID) -> PinnedListing( + row.get(WORKFLOW.PUBLISHED_NAME), + row.get(WORKFLOW.PUBLISHED_DESCRIPTION), + row.get(WORKFLOW.PUBLISHED_DEFAULT_VIEW), + row.get(drifted) + ) + ) + .toMap + } + /** As [[publicCopyOf]], for callers holding only a wid. 404s unless the workflow is public. */ def publicCopyOf(wid: Integer): PublicCopy = { val workflow = requireWorkflow(wid) diff --git a/amber/src/main/scala/org/apache/texera/web/resource/dashboard/user/workflow/WorkflowResource.scala b/amber/src/main/scala/org/apache/texera/web/resource/dashboard/user/workflow/WorkflowResource.scala index 6fbd19bcc2c..74c4b525405 100644 --- a/amber/src/main/scala/org/apache/texera/web/resource/dashboard/user/workflow/WorkflowResource.scala +++ b/amber/src/main/scala/org/apache/texera/web/resource/dashboard/user/workflow/WorkflowResource.scala @@ -122,7 +122,10 @@ object WorkflowResource { ownerName: String, workflow: Workflow, ownerId: Integer, - coverImage: Option[String] + coverImage: Option[String], + // Behind the author's working copy? Listings use it to open the published preview instead of + // the editor -- for the author too, since that is what the entry is showing. + hasUnpublishedChanges: Boolean = false ) case class WorkflowWithPrivilege( diff --git a/amber/src/test/scala/org/apache/texera/web/resource/dashboard/UnifiedResourceSchemaSpec.scala b/amber/src/test/scala/org/apache/texera/web/resource/dashboard/UnifiedResourceSchemaSpec.scala index 27856595794..72c4d9eadea 100644 --- a/amber/src/test/scala/org/apache/texera/web/resource/dashboard/UnifiedResourceSchemaSpec.scala +++ b/amber/src/test/scala/org/apache/texera/web/resource/dashboard/UnifiedResourceSchemaSpec.scala @@ -79,9 +79,10 @@ class UnifiedResourceSchemaSpec extends AnyFlatSpec with Matchers { // Sentinels for the two slots that have no convenient distinct table // column of the right type; every other slot uses a real generated column so - // that all 23 originals render differently from one another. + // that all 28 originals render differently from one another. private val sentinelResourceType: Field[String] = JDSL.inline("s-resource-type") private val sentinelStoragePath: Field[String] = JDSL.inline("s-storage-path") + private val sentinelGranted: Field[java.lang.Boolean] = JDSL.inline(java.lang.Boolean.TRUE) private val sentinelSchema: UnifiedResourceSchema = UnifiedResourceSchema( resourceType = sentinelResourceType, @@ -106,7 +107,12 @@ class UnifiedResourceSchemaSpec extends AnyFlatSpec with Matchers { workflowCoverImage = WORKFLOW_COVER_IMAGE.IMAGE, workflowDefaultView = WORKFLOW.DEFAULT_VIEW, modelFramework = MODEL.FRAMEWORK, - modelFormat = MODEL.FORMAT + modelFormat = MODEL.FORMAT, + workflowHasUnpublishedChanges = WORKFLOW.IS_PUBLIC, + workflowPublishedName = WORKFLOW.PUBLISHED_NAME, + workflowPublishedDescription = WORKFLOW.PUBLISHED_DESCRIPTION, + workflowPublishedDefaultView = WORKFLOW.PUBLISHED_DEFAULT_VIEW, + viewerHasGrantedAccess = sentinelGranted ) // Expected projection, in order: alias -> the original it must be built from. @@ -133,13 +139,18 @@ class UnifiedResourceSchemaSpec extends AnyFlatSpec with Matchers { "workflow_cover_image" -> WORKFLOW_COVER_IMAGE.IMAGE, "workflow_default_view" -> WORKFLOW.DEFAULT_VIEW, "model_framework" -> MODEL.FRAMEWORK, - "model_format" -> MODEL.FORMAT + "model_format" -> MODEL.FORMAT, + "workflow_has_unpublished_changes" -> WORKFLOW.IS_PUBLIC, + "workflow_published_name" -> WORKFLOW.PUBLISHED_NAME, + "workflow_published_description" -> WORKFLOW.PUBLISHED_DESCRIPTION, + "workflow_published_default_view" -> WORKFLOW.PUBLISHED_DEFAULT_VIEW, + "viewer_has_granted_access" -> sentinelGranted ) // -- apply(): the projection ------------------------------------------------ - "apply" should "expose all 23 slots as aliases, in the order the UNION ALL depends on" in { - sentinelSchema.allFields should have size 23 + "apply" should "expose all 28 slots as aliases, in the order the UNION ALL depends on" in { + sentinelSchema.allFields should have size 28 sentinelSchema.allFields.map(_.getName) shouldBe expectedProjection.map(_._1) } @@ -159,7 +170,7 @@ class UnifiedResourceSchemaSpec extends AnyFlatSpec with Matchers { // about datasets still union with one that does: the column count and // types have to line up. val defaults = UnifiedResourceSchema() - defaults.allFields should have size 23 + defaults.allFields should have size 28 val rendered = ctx.renderInlined(JDSL.select(defaults.allFields: _*)) rendered should include("'' as \"resourceType\"") rendered should include("cast(null as timestamp) as \"resourceCreationTime\"") @@ -194,12 +205,12 @@ class UnifiedResourceSchemaSpec extends AnyFlatSpec with Matchers { } it should "collapse the all-defaults projection down to one alias per distinct default" in { - // 23 slots, but only seven structurally distinct default expressions, so the + // 28 slots, but only seven structurally distinct default expressions, so the // de-dup collapses the map to seven entries. Worth pinning because it is // surprising, and because it is what makes the keep-first rule observable at - // all: allFields stays at 23 while the translation map does not. + // all: allFields stays at 28 while the translation map does not. val defaults = UnifiedResourceSchema() - defaults.allFields should have size 23 + defaults.allFields should have size 28 translatedAliases(defaults) shouldBe Seq( "resourceType", // DSL.inline("") "resourceCreationTime", // cast(null as timestamp) @@ -211,7 +222,7 @@ class UnifiedResourceSchemaSpec extends AnyFlatSpec with Matchers { ) } - it should "keep every distinct original when the caller supplies 23 distinct Fields" in { + it should "keep every distinct original when the caller supplies 28 distinct Fields" in { // Nothing to collapse here, which is the control case for the two tests // above: the shrinkage they observe comes from duplicate originals only. translatedAliases(sentinelSchema) shouldBe expectedProjection.map(_._1) @@ -219,7 +230,7 @@ class UnifiedResourceSchemaSpec extends AnyFlatSpec with Matchers { it should "drop exactly the duplicated slots of the production workflow projection" in { val workflowSchema = WorkflowSearchQueryBuilder.mappedResourceSchema - workflowSchema.allFields should have size 23 + workflowSchema.allFields should have size 28 val aliases = translatedAliases(workflowSchema) // `uid` duplicates ownerId (WORKFLOW_OF_USER.UID); the rest are slots the // builder left at their default, and the defaults collide by type. @@ -238,7 +249,7 @@ class UnifiedResourceSchemaSpec extends AnyFlatSpec with Matchers { "jOOQ Field equality" should "be structural, which is what makes the de-dup collapse anything" in { // If jOOQ ever switched to identity equality, translatedFieldSet would keep - // all 23 slots and translateRecord would start reading duplicated columns — + // all 28 slots and translateRecord would start reading duplicated columns — // the tests above would flip, and this one says why. JDSL.cast(null, classOf[Integer]) shouldBe JDSL.cast(null, classOf[Integer]) JDSL.inline("") shouldBe JDSL.inline("") diff --git a/amber/src/test/scala/org/apache/texera/web/resource/dashboard/WorkflowSearchQueryBuilderSpec.scala b/amber/src/test/scala/org/apache/texera/web/resource/dashboard/WorkflowSearchQueryBuilderSpec.scala index f0d8fc63d9a..4b6a8a8937b 100644 --- a/amber/src/test/scala/org/apache/texera/web/resource/dashboard/WorkflowSearchQueryBuilderSpec.scala +++ b/amber/src/test/scala/org/apache/texera/web/resource/dashboard/WorkflowSearchQueryBuilderSpec.scala @@ -62,6 +62,15 @@ class WorkflowSearchQueryBuilderSpec extends AnyFlatSpec with Matchers { // The select lists default_view under its own alias (not carried by the WORKFLOW POJO), // and toEntryImpl reads it back by that alias — the record has to carry the column. private val defaultViewField = WORKFLOW.DEFAULT_VIEW.as("workflow_default_view") + private val driftField = + JDSL.field(JDSL.name("workflow_has_unpublished_changes"), classOf[java.lang.Boolean]) + private val publishedNameField = JDSL.field(JDSL.name("workflow_published_name"), classOf[String]) + private val publishedDefaultViewField = + WORKFLOW.PUBLISHED_DEFAULT_VIEW.as("workflow_published_default_view") + private val publishedDescriptionField = + JDSL.field(JDSL.name("workflow_published_description"), classOf[String]) + private val grantedField = + JDSL.field(JDSL.name("viewer_has_granted_access"), classOf[java.lang.Boolean]) private val ownerUid: Integer = Integer.valueOf(42) private val viewerUid: Integer = Integer.valueOf(43) @@ -79,7 +88,14 @@ class WorkflowSearchQueryBuilderSpec extends AnyFlatSpec with Matchers { uidValue: Integer = ownerUid, privilege: PrivilegeEnum = PrivilegeEnum.WRITE, cover: String = "cover-b64", - defaultView: DefaultViewEnum = DefaultViewEnum.CANVAS + defaultView: DefaultViewEnum = DefaultViewEnum.CANVAS, + hasUnpublishedChanges: java.lang.Boolean = null, + // The rows these tests describe belong to a viewer who was granted access, which is what + // leaves the author's own name and description in place. + grantedAccess: java.lang.Boolean = java.lang.Boolean.TRUE, + publishedName: String = null, + publishedDescription: String = null, + publishedDefaultView: DefaultViewEnum = null ): Record = { val record = ctx.newRecord( WORKFLOW.WID, @@ -89,7 +105,12 @@ class WorkflowSearchQueryBuilderSpec extends AnyFlatSpec with Matchers { WORKFLOW_USER_ACCESS.PRIVILEGE, USER.NAME, coverField, - defaultViewField + defaultViewField, + driftField, + publishedNameField, + publishedDescriptionField, + publishedDefaultViewField, + grantedField ) record.set(WORKFLOW.WID, wid) record.set(WORKFLOW.NAME, "wf-name") @@ -99,12 +120,75 @@ class WorkflowSearchQueryBuilderSpec extends AnyFlatSpec with Matchers { record.set(USER.NAME, "owner-name") record.set(coverField, cover) record.set(defaultViewField, defaultView) + record.set(driftField, hasUnpublishedChanges) + record.set(publishedNameField, publishedName) + record.set(publishedDescriptionField, publishedDescription) + record.set(publishedDefaultViewField, publishedDefaultView) + record.set(grantedField, grantedAccess) record } private def workflowOf(record: Record, uid: Integer): DashboardWorkflow = WorkflowSearchQueryBuilder.toEntryImpl(uid, record).workflow.get + // -- what a viewer without granted access is shown -------------------------- + + "toEntryImpl" should "serve the published name and description to a viewer without granted access" in { + // Publishing pins what the public sees, and a listing is where the public meets a workflow, so + // the listing has to serve the pinned copy too rather than the author's live metadata. + val record = translatedRecord( + grantedAccess = java.lang.Boolean.FALSE, + publishedName = "as-published", + publishedDescription = "described-as-published" + ) + + val workflow = workflowOf(record, viewerUid).workflow + + workflow.getName shouldBe "as-published" + workflow.getDescription shouldBe "described-as-published" + } + + it should "serve the pinned view to a viewer without granted access" in { + // A card that opened the frozen copy in the author's live view would offer a Form View over a + // graph that carries no form; the view travels with the name and the description. + val record = translatedRecord( + grantedAccess = java.lang.Boolean.FALSE, + defaultView = DefaultViewEnum.FORM, + publishedName = "as-published", + publishedDefaultView = DefaultViewEnum.CANVAS + ) + + workflowOf(record, viewerUid).workflow.getDefaultView shouldBe DefaultViewEnum.CANVAS + } + + it should "leave the author's live name and description for a viewer who was granted access" in { + val record = translatedRecord( + grantedAccess = java.lang.Boolean.TRUE, + publishedName = "as-published", + publishedDescription = "described-as-published" + ) + + val workflow = workflowOf(record, ownerUid).workflow + + workflow.getName shouldBe "wf-name" + workflow.getDescription shouldBe "wf-description" + } + + it should "treat an unknown access answer as public, which is the safe direction" in { + // Showing the published copy to someone who turns out to have access is harmless; the reverse + // would hand the author's live metadata to the public. + val record = translatedRecord(grantedAccess = null, publishedName = "as-published") + + workflowOf(record, viewerUid).workflow.getName shouldBe "as-published" + } + + it should "keep the author's name when a public workflow has nothing pinned" in { + // Nothing to substitute, so the row is left as it is rather than blanked. + val record = translatedRecord(grantedAccess = java.lang.Boolean.FALSE, publishedName = null) + + workflowOf(record, viewerUid).workflow.getName shouldBe "wf-name" + } + // -- privilege fallback ----------------------------------------------------- "toEntryImpl" should "fall back to NONE when the workflow privilege is NULL" in { diff --git a/amber/src/test/scala/org/apache/texera/web/resource/dashboard/user/workflow/WorkflowPublishSpec.scala b/amber/src/test/scala/org/apache/texera/web/resource/dashboard/user/workflow/WorkflowPublishSpec.scala index 73306d84e37..23b3a5bafe5 100644 --- a/amber/src/test/scala/org/apache/texera/web/resource/dashboard/user/workflow/WorkflowPublishSpec.scala +++ b/amber/src/test/scala/org/apache/texera/web/resource/dashboard/user/workflow/WorkflowPublishSpec.scala @@ -29,7 +29,10 @@ import org.apache.texera.dao.jooq.generated.tables.daos.{ WorkflowUserAccessDao } import org.apache.texera.dao.jooq.generated.tables.pojos.{User, Workflow, WorkflowUserAccess} +import org.apache.texera.web.resource.dashboard.DashboardResource.SearchQueryParams +import org.apache.texera.web.resource.dashboard.hub.HubResource import org.apache.texera.web.resource.dashboard.user.workflow.WorkflowResource.WorkflowIDs +import org.apache.texera.web.resource.dashboard.{DashboardResource, FulltextSearchQueryUtils} import org.scalatest.BeforeAndAfterAll import org.scalatest.flatspec.AnyFlatSpec import org.scalatest.matchers.should.Matchers @@ -45,7 +48,7 @@ import javax.ws.rs.{BadRequestException, ForbiddenException, NotFoundException} /** * Covers the publish state a workflow can be in -- following the author's latest content, as * publishing has always done, or holding a pinned copy of the version the author froze -- and the - * read paths that decide which of the two a caller is served. + * read paths that decide which of the two a caller is served, listings and search among them. */ class WorkflowPublishSpec extends AnyFlatSpec @@ -85,6 +88,7 @@ class WorkflowPublishSpec override protected def beforeAll(): Unit = { initializeDBAndReplaceDSLContext() + FulltextSearchQueryUtils.usePgroonga = false val userDao = new UserDao(getDSLContext.configuration()) userDao.insert(owner) userDao.insert(stranger) @@ -168,6 +172,40 @@ class WorkflowPublishSpec ) .asInstanceOf[HttpServletRequest] + private def keywords(values: String*): util.ArrayList[String] = { + val list = new util.ArrayList[String]() + values.foreach(list.add) + list + } + + private def keywordIds(values: Integer*): util.ArrayList[Integer] = { + val list = new util.ArrayList[Integer]() + values.foreach(list.add) + list + } + + /** The wids a search returns, which is all any of these tests asks of one. */ + private def searchWids(user: SessionUser, params: SearchQueryParams): List[Integer] = + DashboardResource + .searchAllResources(user, params, includePublic = true) + .results + .flatMap(_.workflow.map(_.workflow.getWid)) + + /** One workflow's listing row, as the given user would see it in a dashboard or on the Hub. */ + private def listingOf(user: SessionUser, wid: Integer) = + DashboardResource + .searchAllResources( + user, + SearchQueryParams(workflowIDs = keywordIds(wid)), + includePublic = true + ) + .results + .flatMap(_.workflow) + .head + + /** Nobody: the hub as an unauthenticated visitor reads it. */ + private def anonymous: SessionUser = new SessionUser(new User()) + private def statusOf(wid: Integer): WorkflowPublishService.PublishStatus = workflowResource.getPublishStatus(wid, ownerSession) @@ -945,4 +983,272 @@ class WorkflowPublishSpec .getSize(util.Arrays.asList(pinned)) .get(pinned) shouldBe publishedContent.length } + + behavior of "hub listings" + + it should "list a pinned workflow on the hub under the name and description it froze" in { + // The hub is the public shelf: everything on it is listed as the public sees it, the author + // included. A pin that this listing did not honour would put the author's live title on that + // shelf, and would leave it disagreeing with the hub's own search about what a workflow is + // called. + val wid = createWorkflow("hub_listing_shows_public_copy") + publishPinned(wid) + relabel(wid, "renamed_after_pinning", "described_after_pinning") + + for (viewer <- Seq(stranger.getUid, owner.getUid)) { + val listed = HubResource.fetchDashboardWorkflowsByWids(Seq(wid), viewer).head.workflow + listed.getName shouldBe "hub_listing_shows_public_copy" + listed.getDescription shouldBe "a workflow" + } + } + + it should "list a following workflow on the hub under the author's latest name and description" in { + val wid = createWorkflow("hub_listing_follows_latest") + workflowResource.makePublic(wid, ownerSession) + relabel(wid, "renamed_while_following", "described_while_following") + + val listed = HubResource.fetchDashboardWorkflowsByWids(Seq(wid), owner.getUid).head.workflow + listed.getName shouldBe "renamed_while_following" + listed.getDescription shouldBe "described_while_following" + } + + it should "tell the hub listing when the copy on show is behind the author's working copy" in { + // The card advertises the pinned copy, so clicking it has to open that copy. Without this flag + // the author's own card would take them to their editor and show them something else -- the + // same signal the search listing carries, on the query the hub builds for itself. + val wid = createWorkflow("hub_listing_reports_drift") + publishPinned(wid) + + def listedDrift(): Boolean = + HubResource.fetchDashboardWorkflowsByWids(Seq(wid), owner.getUid).head.hasUnpublishedChanges + + listedDrift() shouldBe false + edit(wid, editedContent) + listedDrift() shouldBe true + workflowResource.pinLatest(wid, ownerSession) + listedDrift() shouldBe false + } + + it should "clone the author's latest name and description while nothing is pinned" in { + val wid = createWorkflow("clone_follows_latest_metadata") + workflowResource.makePublic(wid, ownerSession) + relabel(wid, "renamed_before_clone", "described_before_clone") + + val cloned = + workflowDao.fetchOneByWid(workflowResource.cloneWorkflow(wid, strangerSession, fakeRequest())) + + cloned.getName shouldBe "renamed_before_clone_clone" + cloned.getDescription shouldBe "described_before_clone" + } + + behavior of "search" + + it should "not match a public workflow on anything that exists only in unpublished edits" in { + // Both halves of what public search indexes -- the words in the graph, and the operators in it -- + // have to stop at the pinned copy, or searching would surface drafts nobody can open. + val wid = createWorkflow("search_ignores_drafts") + publishPinned(wid) + edit( + wid, + """{"operators":[{"operatorType":"SecretDraftOperator"}],"note":"supersecretdraftword"}""" + ) + + val byKeyword = + searchWids(anonymous, SearchQueryParams(keywords = keywords("supersecretdraftword"))) + val byOperator = + searchWids(anonymous, SearchQueryParams(operators = keywords("SecretDraftOperator"))) + + byKeyword should not contain wid + byOperator should not contain wid + } + + it should "match a public workflow on keywords in its published copy" in { + val wid = createWorkflow("search_finds_published") + publishPinned(wid) + edit(wid, editedContent) + + searchWids( + anonymous, + SearchQueryParams(keywords = keywords("content_as_published")) + ) should contain(wid) + } + + it should "match an unpinned public workflow on the author's latest" in { + // Following means the public copy is the working copy, so search must reach it through the same + // public path that a pinned workflow reaches its frozen copy through. + val wid = createWorkflow("search_finds_unpinned_latest") + workflowResource.makePublic(wid, ownerSession) + edit(wid, """{"operators":[],"note":"unpinnedsearchword"}""") + + searchWids( + anonymous, + SearchQueryParams(keywords = keywords("unpinnedsearchword")) + ) should contain(wid) + } + + it should "still match the author's own workflow on their unpublished edits" in { + val wid = createWorkflow("search_finds_own_draft") + publishPinned(wid) + edit(wid, """{"operators":[],"note":"myowndraftword"}""") + + searchWids( + ownerSession, + SearchQueryParams(keywords = keywords("myowndraftword")) + ) should contain(wid) + } + + it should "match a pinned workflow on the name and description the public can see" in { + // The graph was already matched against the pinned copy; the title and description are on public + // show just as much, so matching them against the author's live values gets it wrong in both + // directions at once -- findable by a title nobody has seen, unfindable by the one on screen. + val wid = createWorkflow("aapublicnamesearch") + relabel(wid, "aapublicnamesearch", "aapublicdescriptionsearch") + publishPinned(wid) + relabel(wid, "zzsecretnamesearch", "zzsecretdescriptionsearch") + + def anonymousHits(word: String): List[Integer] = + searchWids(anonymous, SearchQueryParams(keywords = keywords(word))) + + anonymousHits("aapublicnamesearch") should contain(wid) + anonymousHits("aapublicdescriptionsearch") should contain(wid) + anonymousHits("zzsecretnamesearch") should not contain wid + anonymousHits("zzsecretdescriptionsearch") should not contain wid + } + + it should "match an unpinned public workflow on its live name and description" in { + val wid = createWorkflow("bbfollowingnamesearch") + workflowResource.makePublic(wid, ownerSession) + relabel(wid, "bbrenamedwhilefollowing", "a workflow") + + searchWids( + anonymous, + SearchQueryParams(keywords = keywords("bbrenamedwhilefollowing")) + ) should contain(wid) + } + + it should "still match the author's own workflow on a name only they can see" in { + val wid = createWorkflow("ccownnamesearch") + publishPinned(wid) + relabel(wid, "ccprivaterenamesearch", "a workflow") + + searchWids( + ownerSession, + SearchQueryParams(keywords = keywords("ccprivaterenamesearch")) + ) should contain(wid) + } + + it should "show a public viewer the published name and description in a listing" in { + // The listing is where people find and click a workflow, so a title that keeps following the + // author defeats the freeze exactly as an unfrozen graph would: a report about a title could be + // answered by quietly editing the title. + val wid = createWorkflow("listing_name_before") + publishPinned(wid) + relabel(wid, "listing_name_after", "described after publishing") + + def listedAs(session: SessionUser): (String, String) = { + val entry = listingOf(session, wid).workflow + (entry.getName, entry.getDescription) + } + + // The author is not a public viewer of their own workflow. + listedAs(ownerSession) shouldBe ("listing_name_after", "described after publishing") + // A stranger reaches it only because it is public. + listedAs(strangerSession) shouldBe ("listing_name_before", "a workflow") + // ...and the listing now agrees with what opening it shows. + workflowResource.retrievePublicWorkflow(wid).name shouldBe "listing_name_before" + } + + it should "show a public viewer the author's live name while nothing is pinned" in { + val wid = createWorkflow("listing_unpinned_before") + workflowResource.makePublic(wid, ownerSession) + relabel(wid, "listing_unpinned_after", "a workflow") + + listingOf(strangerSession, wid).workflow.getName shouldBe "listing_unpinned_after" + } + + it should "show a public viewer the view the copy on show opens in" in { + // The Form View's definition rides inside the content, so a card offering a form over a frozen + // canvas would open a form the copy does not carry. The view travels with the name and the + // description, and for the author's own row it stays theirs. + val wid = createWorkflow("listing_freezes_the_view") + publishPinned(wid) + getDSLContext + .update(WORKFLOW) + .set(WORKFLOW.DEFAULT_VIEW, DefaultViewEnum.FORM) + .where(WORKFLOW.WID.eq(wid)) + .execute() + + listingOf(strangerSession, wid).workflow.getDefaultView shouldBe DefaultViewEnum.CANVAS + listingOf(ownerSession, wid).workflow.getDefaultView shouldBe DefaultViewEnum.FORM + } + + it should "show a collaborator the author's live name in a listing" in { + val wid = createWorkflow("listing_for_collaborator") + publishPinned(wid) + grantAccess(wid, PrivilegeEnum.READ) + + try { + relabel(wid, "renamed_after_sharing", "a workflow") + listingOf(strangerSession, wid).workflow.getName shouldBe "renamed_after_sharing" + } finally revokeAccess(wid) + } + + it should "leave a private workflow's own listing untouched" in { + val wid = createWorkflow("listing_private") + relabel(wid, "private_renamed", "a workflow") + + listingOf(ownerSession, wid).workflow.getName shouldBe "private_renamed" + } + + it should "tell listings when the copy on show is behind the author's working copy" in { + // What sends the author to the published preview instead of their editor when they click their + // own workflow in the hub: the entry is advertising the pinned version, not what they are editing. + val wid = createWorkflow("listing_reports_drift") + publishPinned(wid) + + def drifted(): Boolean = listingOf(ownerSession, wid).hasUnpublishedChanges + + drifted() shouldBe false + edit(wid, editedContent) + drifted() shouldBe true + workflowResource.pinLatest(wid, ownerSession) + drifted() shouldBe false + } + + it should "report the same drift to a listing as to the share dialog" in { + // Three places answer "is the public behind?": the share dialog in Scala, the search projection + // and the hub listing in SQL. A rename or a change of view exercises the fields most easily left + // out of one of them, and a card that disagrees with the dialog is a card the author distrusts. + val wid = createWorkflow("listing_matches_dialog") + publishPinned(wid) + + relabel(wid, "renamed_after_pinning", "a workflow") + listingOf(ownerSession, wid).hasUnpublishedChanges shouldBe statusOf(wid).hasUnpublishedChanges + listingOf(ownerSession, wid).hasUnpublishedChanges shouldBe true + + workflowResource.pinLatest(wid, ownerSession) + getDSLContext + .update(WORKFLOW) + .set(WORKFLOW.DEFAULT_VIEW, DefaultViewEnum.FORM) + .where(WORKFLOW.WID.eq(wid)) + .execute() + + listingOf(ownerSession, wid).hasUnpublishedChanges shouldBe statusOf(wid).hasUnpublishedChanges + listingOf(ownerSession, wid).hasUnpublishedChanges shouldBe true + } + + it should "never report drift for a workflow with nothing frozen" in { + // Drift is "what the public sees is not what you have", so it can only be true of a workflow that + // has a frozen copy at all. Private and public-but-following both have none. + def driftOf(wid: Integer): Boolean = listingOf(ownerSession, wid).hasUnpublishedChanges + + val priv = createWorkflow("listing_ignores_private") + edit(priv, editedContent) + driftOf(priv) shouldBe false + + val following = createWorkflow("listing_ignores_unpinned") + workflowResource.makePublic(following, ownerSession) + edit(following, editedContent) + driftOf(following) shouldBe false + } } From 43494d3ffafc774ec5bdff7bfc35cc1125dc221a Mon Sep 17 00:00:00 2001 From: Yang Zhang Date: Sat, 22 Aug 2026 14:59:57 -0700 Subject: [PATCH 2/2] feat(workflow): mark the pinned version in the revision history The author can pin a version, and their working copy then moves on. This gives them a way back to what the public is seeing: pinning leaves an anchor in the revision history they already use, and the panel marks it as the one currently public, so restoring the published version is the restore they already know. The anchor is a version row whose delta is the identity patch, so replaying it returns exactly what was pinned however many edits pile up after. An existing row cannot stand in: a version row replays to the content as it was *before* the change it records. Pinning content that is already the pinned one reuses its anchor rather than adding a twin, and the read that decides takes a row lock so two pins racing cannot both insert. Two consequences the anchor forces: - It must not start the version panel's aggregation window. It lands seconds after the save it freezes, and the panel folds close-together versions into the newest, which would hide the author's own save behind a row they never made. - The revision history is now readable only with granted access, or while nothing is pinned. Replaying a version folds deltas back from the author's current content, so listing versions of a pinned workflow would hand a public viewer the very edits the pin is holding back. `publish-status` carries the pinned version's date, read from the version row so the dialog and the panel print one value rather than two clocks'. Co-Authored-By: Claude Opus 5 --- .../workflow/WorkflowPublishService.scala | 70 ++++++++-- .../user/workflow/WorkflowResource.scala | 6 +- .../workflow/WorkflowVersionResource.scala | 48 ++++--- .../user/workflow/WorkflowPublishSpec.scala | 129 ++++++++++++++++-- .../WorkflowVersionResourceSpec.scala | 32 ++++- 5 files changed, 246 insertions(+), 39 deletions(-) diff --git a/amber/src/main/scala/org/apache/texera/web/resource/dashboard/user/workflow/WorkflowPublishService.scala b/amber/src/main/scala/org/apache/texera/web/resource/dashboard/user/workflow/WorkflowPublishService.scala index 62d15fde230..61483f25600 100644 --- a/amber/src/main/scala/org/apache/texera/web/resource/dashboard/user/workflow/WorkflowPublishService.scala +++ b/amber/src/main/scala/org/apache/texera/web/resource/dashboard/user/workflow/WorkflowPublishService.scala @@ -22,7 +22,7 @@ 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 @@ -30,6 +30,8 @@ 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,6 +226,22 @@ 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) @@ -200,6 +249,7 @@ object WorkflowPublishService extends LazyLogging { 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)) diff --git a/amber/src/main/scala/org/apache/texera/web/resource/dashboard/user/workflow/WorkflowResource.scala b/amber/src/main/scala/org/apache/texera/web/resource/dashboard/user/workflow/WorkflowResource.scala index 74c4b525405..e5d71281a67 100644 --- a/amber/src/main/scala/org/apache/texera/web/resource/dashboard/user/workflow/WorkflowResource.scala +++ b/amber/src/main/scala/org/apache/texera/web/resource/dashboard/user/workflow/WorkflowResource.scala @@ -964,14 +964,16 @@ class WorkflowResource extends LazyLogging { .fetchOne() // 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. diff --git a/amber/src/main/scala/org/apache/texera/web/resource/dashboard/user/workflow/WorkflowVersionResource.scala b/amber/src/main/scala/org/apache/texera/web/resource/dashboard/user/workflow/WorkflowVersionResource.scala index c218a1d4ef8..faefabb6393 100644 --- a/amber/src/main/scala/org/apache/texera/web/resource/dashboard/user/workflow/WorkflowVersionResource.scala +++ b/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)) ) } } diff --git a/amber/src/test/scala/org/apache/texera/web/resource/dashboard/user/workflow/WorkflowPublishSpec.scala b/amber/src/test/scala/org/apache/texera/web/resource/dashboard/user/workflow/WorkflowPublishSpec.scala index 23b3a5bafe5..9ba7cbcb0a8 100644 --- a/amber/src/test/scala/org/apache/texera/web/resource/dashboard/user/workflow/WorkflowPublishSpec.scala +++ b/amber/src/test/scala/org/apache/texera/web/resource/dashboard/user/workflow/WorkflowPublishSpec.scala @@ -48,7 +48,8 @@ import javax.ws.rs.{BadRequestException, ForbiddenException, NotFoundException} /** * Covers the publish state a workflow can be in -- following the author's latest content, as * publishing has always done, or holding a pinned copy of the version the author froze -- and the - * read paths that decide which of the two a caller is served, listings and search among them. + * read paths that decide which of the two a caller is served, listings and search among them, and + * the anchor a pin leaves in the revision history. */ class WorkflowPublishSpec extends AnyFlatSpec @@ -144,14 +145,17 @@ class WorkflowPublishSpec private val aWorkflowUpdate = """(?is)\s*update\s+(?:\S*\.)?["`\[]?workflow["`\]]?\s+set\b.*""".r - private def interleaving(interleaved: () => Unit)(act: => Unit): Unit = { + private def interleaving( + interleaved: () => Unit, + at: String => Boolean = aWorkflowUpdate.matches + )(act: => Unit): Unit = { var pending = true val configuration = getDSLContext.configuration().asInstanceOf[DefaultConfiguration] val previousListeners = configuration.executeListenerProviders() configuration.set(new DefaultExecuteListenerProvider(new ExecuteListener { override def executeStart(ctx: ExecuteContext): Unit = { - val sql = Option(ctx.sql()).getOrElse("") - if (pending && aWorkflowUpdate.matches(sql)) { + val sql = Option(ctx.sql()).getOrElse("").toLowerCase + if (pending && at(sql)) { pending = false interleaved() } @@ -160,8 +164,9 @@ class WorkflowPublishSpec try act finally configuration.set(previousListeners: _*) withClue("nothing was interleaved, so this proves nothing: ") { pending shouldBe false } - } /** Clone records the caller's IP, and that is the only thing it wants from the request. - */ + } + + /** Clone records the caller's IP, and that is the only thing it wants from the request. */ private def fakeRequest(): HttpServletRequest = Proxy .newProxyInstance( @@ -330,10 +335,14 @@ class WorkflowPublishSpec // A pin that read the row and wrote what it had read would freeze the version before a save // landing in that window -- and the author, who had just pinned, would be told they have // unpublished changes. Each column is copied from its own row instead, so there is no window. + // + // Interleaved on the way in rather than mid-transaction: pinning now locks the row first, so a + // save arriving after that waits for the pin to finish -- which is the point of the lock. This + // is the last moment a save can still get in front of it. val wid = createWorkflow("pin_takes_the_row_as_it_stands") workflowResource.makePublic(wid, ownerSession) - interleaving(() => edit(wid, editedContent)) { + interleaving(() => edit(wid, editedContent), at = _.contains("for update")) { workflowResource.pinLatest(wid, ownerSession) } @@ -345,11 +354,14 @@ class WorkflowPublishSpec // The "is it published?" check and the pin are two statements. An unpublish landing between // them used to be undone by the pin's own `is_public = true`, putting a workflow the author had // just taken down back on the Hub -- and pinned, so the copy they removed is what the public - // reads. + // reads. Interleaved on the way in: once the pin has the row's lock, an unpublish waits for it. val wid = createWorkflow("pin_cannot_republish") publishPinned(wid) - interleaving(() => workflowResource.makePrivate(wid, ownerSession)) { + interleaving( + () => workflowResource.makePrivate(wid, ownerSession), + at = _.contains("for update") + ) { a[BadRequestException] should be thrownBy workflowResource.pinLatest(wid, ownerSession) } @@ -1251,4 +1263,103 @@ class WorkflowPublishSpec edit(following, editedContent) driftOf(following) shouldBe false } + + behavior of "the revision history" + + it should "leave an anchor in the revision history the author can identify" in { + val wid = createWorkflow("publish_leaves_anchor") + publishPinned(wid) + + val marked = versionResource + .retrieveVersionsOfWorkflow(wid, ownerSession) + .filter(_.isCurrentlyPublic) + marked should have size 1 + marked.head.vId shouldBe workflowDao.fetchOneByWid(wid).getPublishedVersionId + // Both collapsing rules would otherwise hide it, and a hidden anchor is useless to the author. + marked.head.importance shouldBe true + // The revision panel prints this row's creation time and the share dialog prints the pinned + // version's date. They are one row, so they have to be one value: the same version dated a + // second apart in two panels reads as two versions. + marked.head.creationTime shouldBe statusOf(wid).pinnedVersionTime.get + } + + it should "keep the save a pin was taken from visible in the revision panel" in { + // The anchor lands seconds after the save it freezes, and the panel folds versions that land + // close together into the newest of them. Letting the anchor be that newest one would hide the + // author's own save behind a row they never made. + val wid = createWorkflow("pin_does_not_swallow_the_save") + edit(wid, editedContent) + publishPinned(wid) + + val shown = versionResource + .retrieveVersionsOfWorkflow(wid, ownerSession) + .filter(_.importance) + // The anchor, and the save it was taken from. + shown.count(_.isCurrentlyPublic) shouldBe 1 + shown.size should be >= 2 + } + + it should "move the mark when the pin moves, leaving the old anchor unmarked" in { + val wid = createWorkflow("only_the_live_one_is_marked") + publishPinned(wid) + val first = workflowDao.fetchOneByWid(wid).getPublishedVersionId + edit(wid, editedContent) + workflowResource.pinLatest(wid, ownerSession) + val second = workflowDao.fetchOneByWid(wid).getPublishedVersionId + + second should not be first + versionResource + .retrieveVersionsOfWorkflow(wid, ownerSession) + .filter(_.isCurrentlyPublic) + .map(_.vId) shouldBe List(second) + } + + it should "replay the anchor to exactly what was published, however much is edited after" in { + val wid = createWorkflow("anchor_replays_to_published") + publishPinned(wid) + val anchor = workflowDao.fetchOneByWid(wid).getPublishedVersionId + + edit(wid, editedContent) + edit(wid, """{"operators":[],"note":"later_still"}""") + + versionResource + .retrieveWorkflowVersion(wid, anchor, ownerSession) + .getContent shouldBe publishedContent + } + + it should "not record a second anchor when pinning again without having edited" in { + // Otherwise a repeated click would litter the author's revision panel with rows that all replay + // to the same graph. + val wid = createWorkflow("republish_without_edits") + val before = versionResource.retrieveVersionsOfWorkflow(wid, ownerSession).size + publishPinned(wid) + val anchor = workflowDao.fetchOneByWid(wid).getPublishedVersionId + workflowResource.pinLatest(wid, ownerSession) + workflowResource.pinLatest(wid, ownerSession) + + workflowDao.fetchOneByWid(wid).getPublishedVersionId shouldBe anchor + versionResource.retrieveVersionsOfWorkflow(wid, ownerSession) should have size (before + 1) + } + + it should "keep the version the pin was taken from after the pin is dropped" in { + val wid = createWorkflow("unpin_keeps_history") + publishPinned(wid) + + workflowResource.unpin(wid, ownerSession) + + versionResource.retrieveVersionsOfWorkflow(wid, ownerSession) should not be empty + workflowDao.fetchOneByWid(wid).getPublishedVersionId shouldBe null + } + + it should "date the public view by the version on show" in { + // A public viewer is told when what they are looking at was published, not when the author last + // touched a copy they cannot see. + val wid = createWorkflow("public_view_is_dated_by_the_pin") + publishPinned(wid) + val pinnedAt = statusOf(wid).pinnedVersionTime.get + + edit(wid, editedContent) + + workflowResource.retrievePublicWorkflow(wid).lastModifiedTime shouldBe pinnedAt + } } diff --git a/amber/src/test/scala/org/apache/texera/web/resource/dashboard/user/workflow/WorkflowVersionResourceSpec.scala b/amber/src/test/scala/org/apache/texera/web/resource/dashboard/user/workflow/WorkflowVersionResourceSpec.scala index 91b0b1c1be0..35475ec9ca3 100644 --- a/amber/src/test/scala/org/apache/texera/web/resource/dashboard/user/workflow/WorkflowVersionResourceSpec.scala +++ b/amber/src/test/scala/org/apache/texera/web/resource/dashboard/user/workflow/WorkflowVersionResourceSpec.scala @@ -551,10 +551,40 @@ class WorkflowVersionResourceSpec version(3, 0L, positional), // latest — important regardless of content version(2, TimeUnit.SECONDS.toMillis(1), meaningful), // within the aggregate window version(1, TimeUnit.DAYS.toMillis(30), meaningful) // outside it, judged on content - ) + ), + Option.empty[Integer] ) encoded.map(_.vId) shouldBe List(3, 2, 1) encoded.map(_.importance) shouldBe List(true, false, true) + encoded.map(_.isCurrentlyPublic) shouldBe List(false, false, false) + } + + it should "keep the version on public show visible even when the rules would collapse it" in { + val base = 1_700_000_000_000L + def version(vid: Int, offsetMillis: Long, content: String): WorkflowVersion = { + val v = new WorkflowVersion + v.setVid(vid) + v.setWid(testWorkflowWid) + v.setContent(content) + v.setCreationTime(new Timestamp(base - offsetMillis)) + v + } + + val positional = """[{"op":"replace","path":"/operatorPositions/op-1/x","value":5}]""" + + // Version 2 lands inside the aggregation window and carries a positional-only + // patch, so both collapsing rules would hide it. Being the published one wins. + val encoded = WorkflowVersionResource invokePrivate encodeVersionImportance( + List( + version(3, 0L, positional), + version(2, TimeUnit.SECONDS.toMillis(1), positional), + version(1, TimeUnit.SECONDS.toMillis(2), positional) + ), + Option(Integer.valueOf(2)) + ) + + encoded.map(_.importance) shouldBe List(true, true, false) + encoded.map(_.isCurrentlyPublic) shouldBe List(false, true, false) } }