refact(server): create new list and lookup worker queries

Author: irenarindos

internal/daemon/controller/handlers/workers/worker_service.go

@@ -900,7 +900,7 @@ func (s Service) toProto(ctx context.Context, in *server.Worker, opt ...handlers
 		out.LastStatusTime = in.GetLastStatusTime().GetTimestamp()
 	}
 	if outputFields.Has(globals.ActiveConnectionCountField) {
-		out.ActiveConnectionCount = &wrapperspb.UInt32Value{Value: in.ActiveConnectionCount()}
+		out.ActiveConnectionCount = &wrapperspb.UInt32Value{Value: in.ActiveConnectionCount}
 	}
 	if outputFields.Has(globals.ControllerGeneratedActivationToken) && in.ControllerGeneratedActivationToken != "" {
 		out.ControllerGeneratedActivationToken = &wrapperspb.StringValue{Value: in.ControllerGeneratedActivationToken}

internal/db/schema/migrations/oss/postgres/86/01_server_worker_local_storage_state.up.sql

@@ -36,7 +36,7 @@ alter table server_worker
       on update cascade;
 
 drop view server_worker_aggregate;
--- Updated in 99/01_split_worker_tag_tables.up.sql
+-- Removed in 94/01_split_worker_tag_tables.up.sql
 -- Replaces view created in 52/01_worker_operational_state.up.sql to add the worker local storage state
 create view server_worker_aggregate as
 with worker_config_tags(worker_id, source, tags) as (

internal/db/schema/migrations/oss/postgres/94/01_split_worker_tag_table.up.sql

@@ -0,0 +1,54 @@
+-- Copyright (c) HashiCorp, Inc.
+-- SPDX-License-Identifier: BUSL-1.1
+
+begin;
+
+-- Create new tables for worker config and worker api tags
+create table server_worker_config_tag(
+  worker_id wt_public_id
+    constraint server_worker_fkey
+      references server_worker (public_id)
+      on delete cascade
+      on update cascade,
+  key wt_tagpair,
+  value wt_tagpair,
+  primary key (worker_id, key, value)
+);
+comment on table server_worker_config_tag is
+  'server_worker_config_tag is a table where each row represents a worker config tag.';
+
+create table server_worker_api_tag(
+  worker_id wt_public_id
+    constraint server_worker_fkey
+      references server_worker (public_id)
+      on delete cascade
+      on update cascade,
+  key wt_tagpair,
+  value wt_tagpair,
+  primary key (worker_id, key, value)
+);
+comment on table server_worker_api_tag is
+  'server_worker_api_tag is a table where each row represents a worker api tag.';
+
+-- Migrate from server_worker_tag to the new tables
+ insert into server_worker_config_tag
+             (worker_id, key, value)
+      select worker_id, key, value
+        from server_worker_tag
+       where source = 'configuration';
+
+ insert into server_worker_api_tag
+             (worker_id, key, value)
+      select worker_id, key, value
+        from server_worker_tag
+       where source = 'api';
+
+-- Removes view created in 86/01_server_worker_local_storage_state.up.sql
+-- This view is removed in favor of custom sql in query.go
+drop view server_worker_aggregate;
+
+-- Drop the old tables
+drop table server_worker_tag;
+drop table server_worker_tag_enm;
+
+commit;

internal/db/schema/migrations/oss/postgres/99/01_split_worker_tag_table.up.sql

@@ -29,7 +29,7 @@ func Test_WorkerTagTableSplit(t *testing.T) {
 	t.Parallel()
 	require := require.New(t)
 	const priorMigration = 92001
-	const serverEnumMigration = 99001
+	const serverEnumMigration = 94001
 	dialect := dbtest.Postgres
 	ctx := context.Background()
 

internal/db/schema/migrations/oss/postgres_99_01_test.go renamed to internal/db/schema/migrations/oss/postgres_94_01_test.go

@@ -4,6 +4,82 @@
 package server
 
 const (
+	listWorkersQuery = `
+	with connection_count (worker_id, count) as (
+    select worker_id,
+           count(1) as count
+      from session_connection
+     where closed_reason is null
+  group by worker_id
+  )
+    select w.public_id,
+           w.scope_id,
+           w.description,
+           w.name,
+           w.address,
+           w.create_time,
+           w.update_time,
+           w.version,
+           w.last_status_time,
+           w.type,
+           w.release_version,
+           w.operational_state,
+           w.local_storage_state,
+           cc.count as active_connection_count,
+           wt.tags as api_tags,
+           ct.tags as config_tags
+      from server_worker w
+ left join (   select worker_id, 
+                     json_agg(json_build_object('key', key, 'value', value)) as tags 
+                from server_worker_api_tag 
+               group by worker_id) wt
+        on w.public_id = wt.worker_id
+ left join (   select worker_id, 
+                     json_agg(json_build_object('key', key, 'value', value)) as tags 
+                from server_worker_config_tag group by worker_id) ct
+        on w.public_id = ct.worker_id
+ left join connection_count as cc
+        on w.public_id = cc.worker_id
+    `
+
+	lookupWorkerQuery = `
+	with connection_count (worker_id, count) as (
+    select worker_id,
+           count(1) as count
+      from session_connection
+     where closed_reason is null
+  group by worker_id
+  )
+    select w.public_id,
+           w.scope_id,
+           w.description,
+           w.name,
+           w.address,
+           w.create_time,
+           w.update_time,
+           w.version,
+           w.last_status_time,
+           w.type,
+           w.release_version,
+           w.operational_state,
+           w.local_storage_state,
+           cc.count as active_connection_count,
+           wt.tags as api_tags,
+           ct.tags as config_tags
+      from server_worker w
+ left join (   select worker_id, 
+                     json_agg(json_build_object('key', key, 'value', value)) as tags 
+                from server_worker_api_tag 
+               group by worker_id) wt
+        on w.public_id = wt.worker_id
+ left join (   select worker_id, 
+                     json_agg(json_build_object('key', key, 'value', value)) as tags 
+                from server_worker_config_tag group by worker_id) ct
+        on w.public_id = ct.worker_id
+ left join connection_count as cc
+        on w.public_id = cc.worker_id
+     where w.public_id = @worker_id
+    `
 	getStorageBucketCredentialStatesByWorkerId = `
 		select spsb.public_id as storage_bucket_id,
 			   wsbcs.permission_type, wsbcs.state,