feat(domain multi-tenancy): Update persistence layer to use the new fields in transfer task and timer task (#7729)

<!-- 1-2 line summary of WHAT changed technically:
- Always link the relevant projects GitHub issue, unless it is a minor
bugfix
- Good: "Modified FailoverDomain mapper to allow null ActiveClusterName
#320"
- Bad: "added nil check" -->
**What changed?**

Update persistence layer to store the new fields of transfer task and
timer task into the underlying database

<!-- Your goal is to provide all the required context for a future
maintainer
to understand the reasons for making this change (see
https://cbea.ms/git-commit/#why-not-how).
How did this work previously (and what was wrong with it)? What has
changed, and why did you solve it
this way?
- Good: "Active-active domains have independent cluster attributes per
region. Previously,
modifying cluster attributes required spedifying the default
ActiveClusterName which
updates the global domain default. This prevents operators from updating
regional
configurations without affecting the primary cluster designation. This
change allows
  attribute updates to be independent of active cluster selection."
- Bad: "Improves domain handling" -->
**Why?**
To support Domain multi-tenancy, we want all history tasks to be tagged
by their task lists.
#7724

<!-- Include specific test commands and setup. Please include the exact
commands such that
another maintainer or contributor can reproduce the test steps taken. 
- e.g Unit test commands with exact invocation
`go test -v ./common/types/mapper/proto -run TestFailoverDomainRequest`
- For integration tests include setup steps and test commands
Example: "Started local server with `./cadence start`, then ran `make
test_e2e`"
- For local simulation testing include setup steps for the server and
how you ran the tests
- Good: Full commands that reviewers can copy-paste to verify
- Bad: "Tested locally" or "Added tests" -->
**How did you test it?**
unit tests 
```
cd common/persistence && go test ./...
```

<!-- If there are risks that the release engineer should know about
document them here.
For example:
- Has an API/IDL been modified? Is it backwards/forwards compatible? If
not, what are the repecussions?
- Has a schema change been introduced? Is it possible to roll back?
- Has a feature flag been re-used for a new purpose? 
- Is there a potential performance concern? Is the change modifying core
task processing logic?
- If truly N/A, you can mark it as such -->
**Potential risks**
N/A

<!-- If this PR completes a user facing feature or changes functionality
add release notes here.
Your release notes should allow a user and the release engineer to
understand the changes with little context.
Always ensure that the description contains a link to the relevant
GitHub issue. -->
**Release notes**


<!-- Consider whether this change requires documentation updates in the
Cadence-Docs repo
- If yes: mention what needs updating (or link to docs PR in
cadence-docs repo)
- If in doubt, add a note about potential doc needs
- Only mark N/A if you're certain no docs are affected -->
**Documentation Changes**


---

## Reviewer Validation

**PR Description Quality** (check these before reviewing code):

- [ ] **"What changed"** provides a clear 1-2 line summary
  - [ ] Project Issue is linked
- [ ] **"Why"** explains the full motivation with sufficient context
- [ ] **Testing is documented:**
- [ ] Unit test commands are included (with exact `go test` invocation)
- [ ] Integration test setup/commands included (if integration tests
were run)
  - [ ] Canary testing details included (if canary was mentioned)
- [ ] **Potential risks** section is thoughtfully filled out (or
legitimately N/A)
- [ ] **Release notes** included if this completes a user-facing feature
- [ ] **Documentation** needs are addressed (or noted if uncertain)

Author: Shaddoll

common/persistence/data_manager_interfaces.go

@@ -487,6 +487,7 @@ type (
 		ScheduleID              int64
 		Version                 int64
 		RecordVisibility        bool
+		OriginalTaskList        string
 	}
 
 	// CrossClusterTaskInfo describes a cross-cluster task
@@ -524,6 +525,7 @@ type (
 		EventID             int64
 		ScheduleAttempt     int64
 		Version             int64
+		TaskList            string
 	}
 
 	// TaskListInfo describes a state of a task list implementation.
@@ -1851,6 +1853,16 @@ func (t *TransferTaskInfo) GetDomainID() string {
 	return t.DomainID
 }
 
+// GetTaskList returns the task list for transfer task
+func (t *TransferTaskInfo) GetTaskList() string {
+	return t.TaskList
+}
+
+// GetOriginalTaskList returns the original task list for transfer task
+func (t *TransferTaskInfo) GetOriginalTaskList() string {
+	return t.OriginalTaskList
+}
+
 // String returns a string representation for transfer task
 func (t *TransferTaskInfo) String() string {
 	return fmt.Sprintf("%#v", t)
@@ -1883,39 +1895,50 @@ func (t *TransferTaskInfo) ToTask() (Task, error) {
 			TargetDomainID:     t.TargetDomainID,
 			TaskList:           t.TaskList,
 			ScheduleID:         t.ScheduleID,
+			OriginalTaskList:   t.OriginalTaskList,
 		}, nil
 	case TransferTaskTypeCloseExecution:
 		return &CloseExecutionTask{
 			WorkflowIdentifier: workflowIdentifier,
 			TaskData:           taskData,
+			TaskList:           t.TaskList,
 		}, nil
 	case TransferTaskTypeRecordWorkflowStarted:
 		return &RecordWorkflowStartedTask{
 			WorkflowIdentifier: workflowIdentifier,
 			TaskData:           taskData,
+			TaskList:           t.TaskList,
 		}, nil
 	case TransferTaskTypeResetWorkflow:
 		return &ResetWorkflowTask{
 			WorkflowIdentifier: workflowIdentifier,
 			TaskData:           taskData,
+			TaskList:           t.TaskList,
 		}, nil
 	case TransferTaskTypeRecordWorkflowClosed:
 		return &RecordWorkflowClosedTask{
 			WorkflowIdentifier: workflowIdentifier,
 			TaskData:           taskData,
+			TaskList:           t.TaskList,
 		}, nil
 	case TransferTaskTypeRecordChildExecutionCompleted:
+		targetRunID := t.TargetRunID
+		if t.TargetRunID == TransferTaskTransferTargetRunID {
+			targetRunID = ""
+		}
 		return &RecordChildExecutionCompletedTask{
 			WorkflowIdentifier: workflowIdentifier,
 			TaskData:           taskData,
 			TargetDomainID:     t.TargetDomainID,
 			TargetWorkflowID:   t.TargetWorkflowID,
-			TargetRunID:        t.TargetRunID,
+			TargetRunID:        targetRunID,
+			TaskList:           t.TaskList,
 		}, nil
 	case TransferTaskTypeUpsertWorkflowSearchAttributes:
 		return &UpsertWorkflowSearchAttributesTask{
 			WorkflowIdentifier: workflowIdentifier,
 			TaskData:           taskData,
+			TaskList:           t.TaskList,
 		}, nil
 	case TransferTaskTypeStartChildExecution:
 		return &StartChildExecutionTask{
@@ -1924,26 +1947,37 @@ func (t *TransferTaskInfo) ToTask() (Task, error) {
 			TargetDomainID:     t.TargetDomainID,
 			TargetWorkflowID:   t.TargetWorkflowID,
 			InitiatedID:        t.ScheduleID,
+			TaskList:           t.TaskList,
 		}, nil
 	case TransferTaskTypeCancelExecution:
+		targetRunID := t.TargetRunID
+		if t.TargetRunID == TransferTaskTransferTargetRunID {
+			targetRunID = ""
+		}
 		return &CancelExecutionTask{
 			WorkflowIdentifier:      workflowIdentifier,
 			TaskData:                taskData,
 			TargetDomainID:          t.TargetDomainID,
 			TargetWorkflowID:        t.TargetWorkflowID,
-			TargetRunID:             t.TargetRunID,
+			TargetRunID:             targetRunID,
 			InitiatedID:             t.ScheduleID,
 			TargetChildWorkflowOnly: t.TargetChildWorkflowOnly,
+			TaskList:                t.TaskList,
 		}, nil
 	case TransferTaskTypeSignalExecution:
+		targetRunID := t.TargetRunID
+		if t.TargetRunID == TransferTaskTransferTargetRunID {
+			targetRunID = ""
+		}
 		return &SignalExecutionTask{
 			WorkflowIdentifier:      workflowIdentifier,
 			TaskData:                taskData,
 			TargetDomainID:          t.TargetDomainID,
 			TargetWorkflowID:        t.TargetWorkflowID,
-			TargetRunID:             t.TargetRunID,
+			TargetRunID:             targetRunID,
 			InitiatedID:             t.ScheduleID,
 			TargetChildWorkflowOnly: t.TargetChildWorkflowOnly,
+			TaskList:                t.TaskList,
 		}, nil
 	default:
 		return nil, fmt.Errorf("unknown task type: %d", t.TaskType)
@@ -2027,6 +2061,11 @@ func (t *TimerTaskInfo) GetDomainID() string {
 	return t.DomainID
 }
 
+// GetTaskList returns the task list for timer task
+func (t *TimerTaskInfo) GetTaskList() string {
+	return t.TaskList
+}
+
 // String returns a string representation for timer task
 func (t *TimerTaskInfo) String() string {
 	return fmt.Sprintf(
@@ -2054,6 +2093,7 @@ func (t *TimerTaskInfo) ToTask() (Task, error) {
 			EventID:            t.EventID,
 			ScheduleAttempt:    t.ScheduleAttempt,
 			TimeoutType:        t.TimeoutType,
+			TaskList:           t.TaskList,
 		}, nil
 	case TaskTypeActivityTimeout:
 		return &ActivityTimeoutTask{
@@ -2062,35 +2102,41 @@ func (t *TimerTaskInfo) ToTask() (Task, error) {
 			TimeoutType:        t.TimeoutType,
 			EventID:            t.EventID,
 			Attempt:            t.ScheduleAttempt,
+			TaskList:           t.TaskList,
 		}, nil
 	case TaskTypeDeleteHistoryEvent:
 		return &DeleteHistoryEventTask{
 			WorkflowIdentifier: workflowIdentifier,
 			TaskData:           taskData,
+			TaskList:           t.TaskList,
 		}, nil
 	case TaskTypeWorkflowTimeout:
 		return &WorkflowTimeoutTask{
 			WorkflowIdentifier: workflowIdentifier,
 			TaskData:           taskData,
+			TaskList:           t.TaskList,
 		}, nil
 	case TaskTypeUserTimer:
 		return &UserTimerTask{
 			WorkflowIdentifier: workflowIdentifier,
 			TaskData:           taskData,
 			EventID:            t.EventID,
+			TaskList:           t.TaskList,
 		}, nil
 	case TaskTypeActivityRetryTimer:
 		return &ActivityRetryTimerTask{
 			WorkflowIdentifier: workflowIdentifier,
 			TaskData:           taskData,
 			EventID:            t.EventID,
 			Attempt:            t.ScheduleAttempt,
+			TaskList:           t.TaskList,
 		}, nil
 	case TaskTypeWorkflowBackoffTimer:
 		return &WorkflowBackoffTimerTask{
 			WorkflowIdentifier: workflowIdentifier,
 			TaskData:           taskData,
 			TimeoutType:        t.TimeoutType,
+			TaskList:           t.TaskList,
 		}, nil
 	default:
 		return nil, fmt.Errorf("unknown task type: %d", t.TaskType)

common/persistence/data_manager_interfaces_mock.go

@@ -219,57 +219,9 @@ func (d *nosqlExecutionStore) prepareTimerTasksForWorkflowTxn(domainID, workflow
 	var tasks []*nosqlplugin.HistoryMigrationTask
 
 	for _, task := range timerTasks {
-		var eventID int64
-		var attempt int64
-
-		timeoutType := 0
-
-		switch t := task.(type) {
-		case *persistence.DecisionTimeoutTask:
-			eventID = t.EventID
-			timeoutType = t.TimeoutType
-			attempt = t.ScheduleAttempt
-
-		case *persistence.ActivityTimeoutTask:
-			eventID = t.EventID
-			timeoutType = t.TimeoutType
-			attempt = t.Attempt
-
-		case *persistence.UserTimerTask:
-			eventID = t.EventID
-
-		case *persistence.ActivityRetryTimerTask:
-			eventID = t.EventID
-			attempt = int64(t.Attempt)
-
-		case *persistence.WorkflowBackoffTimerTask:
-			timeoutType = t.TimeoutType
-
-		case *persistence.WorkflowTimeoutTask:
-			// noop
-
-		case *persistence.DeleteHistoryEventTask:
-			// noop
-
-		default:
-			return nil, &types.InternalServiceError{
-				Message: fmt.Sprintf("Unknow timer type: %v", task.GetTaskType()),
-			}
-		}
-
-		nt := &nosqlplugin.TimerTask{
-			TaskType:   task.GetTaskType(),
-			DomainID:   domainID,
-			WorkflowID: workflowID,
-			RunID:      runID,
-
-			VisibilityTimestamp: task.GetVisibilityTimestamp(),
-			TaskID:              task.GetTaskID(),
-
-			TimeoutType:     timeoutType,
-			EventID:         eventID,
-			ScheduleAttempt: attempt,
-			Version:         task.GetVersion(),
+		nt, err := task.ToTimerTaskInfo()
+		if err != nil {
+			return nil, err
 		}
 		var blob *persistence.DataBlob
 		if d.dc.EnableHistoryTaskDualWriteMode() {
@@ -387,85 +339,9 @@ func (d *nosqlExecutionStore) prepareTransferTasksForWorkflowTxn(domainID, workf
 	var tasks []*nosqlplugin.HistoryMigrationTask
 
 	for _, task := range transferTasks {
-		var taskList string
-		var scheduleID int64
-		targetDomainID := domainID
-		targetDomainIDs := map[string]struct{}{}
-		targetWorkflowID := persistence.TransferTaskTransferTargetWorkflowID
-		targetRunID := persistence.TransferTaskTransferTargetRunID
-		targetChildWorkflowOnly := false
-
-		switch task.GetTaskType() {
-		case persistence.TransferTaskTypeActivityTask:
-			targetDomainID = task.(*persistence.ActivityTask).TargetDomainID
-			taskList = task.(*persistence.ActivityTask).TaskList
-			scheduleID = task.(*persistence.ActivityTask).ScheduleID
-
-		case persistence.TransferTaskTypeDecisionTask:
-			targetDomainID = task.(*persistence.DecisionTask).TargetDomainID
-			taskList = task.(*persistence.DecisionTask).TaskList
-			scheduleID = task.(*persistence.DecisionTask).ScheduleID
-
-		case persistence.TransferTaskTypeCancelExecution:
-			targetDomainID = task.(*persistence.CancelExecutionTask).TargetDomainID
-			targetWorkflowID = task.(*persistence.CancelExecutionTask).TargetWorkflowID
-			targetRunID = task.(*persistence.CancelExecutionTask).TargetRunID
-			if targetRunID == "" {
-				targetRunID = persistence.TransferTaskTransferTargetRunID
-			}
-			targetChildWorkflowOnly = task.(*persistence.CancelExecutionTask).TargetChildWorkflowOnly
-			scheduleID = task.(*persistence.CancelExecutionTask).InitiatedID
-
-		case persistence.TransferTaskTypeSignalExecution:
-			targetDomainID = task.(*persistence.SignalExecutionTask).TargetDomainID
-			targetWorkflowID = task.(*persistence.SignalExecutionTask).TargetWorkflowID
-			targetRunID = task.(*persistence.SignalExecutionTask).TargetRunID
-			if targetRunID == "" {
-				targetRunID = persistence.TransferTaskTransferTargetRunID
-			}
-			targetChildWorkflowOnly = task.(*persistence.SignalExecutionTask).TargetChildWorkflowOnly
-			scheduleID = task.(*persistence.SignalExecutionTask).InitiatedID
-
-		case persistence.TransferTaskTypeStartChildExecution:
-			targetDomainID = task.(*persistence.StartChildExecutionTask).TargetDomainID
-			targetWorkflowID = task.(*persistence.StartChildExecutionTask).TargetWorkflowID
-			scheduleID = task.(*persistence.StartChildExecutionTask).InitiatedID
-
-		case persistence.TransferTaskTypeRecordChildExecutionCompleted:
-			targetDomainID = task.(*persistence.RecordChildExecutionCompletedTask).TargetDomainID
-			targetWorkflowID = task.(*persistence.RecordChildExecutionCompletedTask).TargetWorkflowID
-			targetRunID = task.(*persistence.RecordChildExecutionCompletedTask).TargetRunID
-			if targetRunID == "" {
-				targetRunID = persistence.TransferTaskTransferTargetRunID
-			}
-
-		case persistence.TransferTaskTypeCloseExecution,
-			persistence.TransferTaskTypeRecordWorkflowStarted,
-			persistence.TransferTaskTypeResetWorkflow,
-			persistence.TransferTaskTypeUpsertWorkflowSearchAttributes,
-			persistence.TransferTaskTypeRecordWorkflowClosed:
-			// No explicit property needs to be set
-
-		default:
-			return nil, &types.InternalServiceError{
-				Message: fmt.Sprintf("Unknown transfer type: %v", task.GetTaskType()),
-			}
-		}
-		t := &nosqlplugin.TransferTask{
-			TaskType:                task.GetTaskType(),
-			DomainID:                domainID,
-			WorkflowID:              workflowID,
-			RunID:                   runID,
-			VisibilityTimestamp:     task.GetVisibilityTimestamp(),
-			TaskID:                  task.GetTaskID(),
-			TargetDomainID:          targetDomainID,
-			TargetDomainIDs:         targetDomainIDs,
-			TargetWorkflowID:        targetWorkflowID,
-			TargetRunID:             targetRunID,
-			TargetChildWorkflowOnly: targetChildWorkflowOnly,
-			TaskList:                taskList,
-			ScheduleID:              scheduleID,
-			Version:                 task.GetVersion(),
+		t, err := task.ToTransferTaskInfo()
+		if err != nil {
+			return nil, err
 		}
 		var blob *persistence.DataBlob
 		if d.dc.EnableHistoryTaskDualWriteMode() {