Update hashtag following

app/Http/Controllers/Api/ApiV1Controller.php

@@ -71,6 +71,7 @@
 	CollectionService,
 	FollowerService,
 	HashtagService,
+	HashtagFollowService,
 	HomeTimelineService,
 	InstanceService,
 	LikeService,
@@ -3780,6 +3781,7 @@ public function followHashtag(Request $request, $id)
 		);
 
 		HashtagService::follow($pid, $tag->id);
+		HashtagFollowService::add($tag->id, $pid);
 
 		return response()->json(FollowedTagResource::make($follows)->toArray($request));
 	}
@@ -3819,6 +3821,7 @@ public function unfollowHashtag(Request $request, $id)
 
 		if($follows) {
 			HashtagService::unfollow($pid, $tag->id);
+			HashtagFollowService::unfollow($tag->id, $pid);
 			$follows->delete();
 		}
 

app/Jobs/HomeFeedPipeline/FeedRemovePipeline.php

@@ -11,6 +11,7 @@
 use Illuminate\Queue\Middleware\WithoutOverlapping;
 use Illuminate\Contracts\Queue\ShouldBeUniqueUntilProcessing;
 use App\Services\FollowerService;
+use App\Services\StatusService;
 use App\Services\HomeTimelineService;
 
 class FeedRemovePipeline implements ShouldQueue, ShouldBeUniqueUntilProcessing
@@ -66,6 +67,8 @@ public function handle(): void
     {
         $ids = FollowerService::localFollowerIds($this->pid);
 
+        HomeTimelineService::rem($this->pid, $this->sid);
+
         foreach($ids as $id) {
             HomeTimelineService::rem($id, $this->sid);
         }

app/Jobs/HomeFeedPipeline/HashtagUnfollowPipeline.php

@@ -0,0 +1,97 @@
+<?php
+
+namespace App\Jobs\HomeFeedPipeline;
+
+use Illuminate\Bus\Queueable;
+use Illuminate\Contracts\Queue\ShouldBeUnique;
+use Illuminate\Contracts\Queue\ShouldQueue;
+use Illuminate\Foundation\Bus\Dispatchable;
+use Illuminate\Queue\InteractsWithQueue;
+use Illuminate\Queue\SerializesModels;
+use Illuminate\Queue\Middleware\WithoutOverlapping;
+use Illuminate\Contracts\Queue\ShouldBeUniqueUntilProcessing;
+use Illuminate\Support\Facades\Cache;
+use App\Follower;
+use App\Hashtag;
+use App\StatusHashtag;
+use App\Services\HashtagFollowService;
+use App\Services\StatusService;
+use App\Services\HomeTimelineService;
+
+class HashtagUnfollowPipeline implements ShouldQueue, ShouldBeUniqueUntilProcessing
+{
+    use Dispatchable, InteractsWithQueue, Queueable, SerializesModels;
+
+    protected $pid;
+    protected $hid;
+
+    public $timeout = 900;
+    public $tries = 3;
+    public $maxExceptions = 1;
+    public $failOnTimeout = true;
+
+    /**
+     * The number of seconds after which the job's unique lock will be released.
+     *
+     * @var int
+     */
+    public $uniqueFor = 3600;
+
+    /**
+     * Get the unique ID for the job.
+     */
+    public function uniqueId(): string
+    {
+        return 'hfp:hashtag:unfollow:' . $this->hid . ':' . $this->pid;
+    }
+
+    /**
+     * Get the middleware the job should pass through.
+     *
+     * @return array<int, object>
+     */
+    public function middleware(): array
+    {
+        return [(new WithoutOverlapping("hfp:hashtag:unfollow:{$this->hid}:{$this->pid}"))->shared()->dontRelease()];
+    }
+
+    /**
+     * Create a new job instance.
+     */
+    public function __construct($hid, $pid)
+    {
+        $this->hid = $hid;
+        $this->pid = $pid;
+    }
+
+    /**
+     * Execute the job.
+     */
+    public function handle(): void
+    {
+        $hid = $this->hid;
+        $pid = $this->pid;
+
+        $statusIds = HomeTimelineService::get($pid, 0, -1);
+
+        if(!$statusIds || !count($statusIds)) {
+            return;
+        }
+
+        $followingIds = Cache::remember('profile:following:'.$pid, 1209600, function() use($pid) {
+            $following = Follower::whereProfileId($pid)->pluck('following_id');
+            return $following->push($pid)->toArray();
+        });
+
+        foreach($statusIds as $id) {
+            $status = StatusService::get($id, false);
+            if(!$status) {
+                HomeTimelineService::rem($pid, $id);
+                continue;
+            }
+            if(!in_array($status['account']['id'], $followingIds)) {
+                HomeTimelineService::rem($pid, $id);
+            }
+        }
+    }
+}

app/Observers/HashtagFollowObserver.php

@@ -0,0 +1,53 @@
+<?php
+
+namespace App\Observers;
+
+use App\HashtagFollow;
+use App\Services\HashtagFollowService;
+use App\Jobs\HomeFeedPipeline\HashtagUnfollowPipeline;
+use Illuminate\Contracts\Events\ShouldHandleEventsAfterCommit;
+
+class HashtagFollowObserver implements ShouldHandleEventsAfterCommit
+{
+    /**
+     * Handle the HashtagFollow "created" event.
+     */
+    public function created(HashtagFollow $hashtagFollow): void
+    {
+        HashtagFollowService::add($hashtagFollow->hashtag_id, $hashtagFollow->profile_id);
+    }
+
+    /**
+     * Handle the HashtagFollow "updated" event.
+     */
+    public function updated(HashtagFollow $hashtagFollow): void
+    {
+    	//
+    }
+
+    /**
+     * Handle the HashtagFollow "deleting" event.
+     */
+    public function deleting(HashtagFollow $hashtagFollow): void
+    {
+        HashtagFollowService::unfollow($hashtagFollow->hashtag_id, $hashtagFollow->profile_id);
+        HashtagUnfollowPipeline::dispatch($hashtagFollow->hashtag_id, $hashtagFollow->profile_id);
+    }
+
+    /**
+     * Handle the HashtagFollow "restored" event.
+     */
+    public function restored(HashtagFollow $hashtagFollow): void
+    {
+        //
+    }
+
+    /**
+     * Handle the HashtagFollow "force deleted" event.
+     */
+    public function forceDeleted(HashtagFollow $hashtagFollow): void
+    {
+        HashtagFollowService::unfollow($hashtagFollow->hashtag_id, $hashtagFollow->profile_id);
+        HashtagUnfollowPipeline::dispatch($hashtagFollow->hashtag_id, $hashtagFollow->profile_id);
+    }
+}

app/Observers/StatusHashtagObserver.php

@@ -38,12 +38,12 @@ public function updated(StatusHashtag $hashtag)
     }
 
     /**
-     * Handle the notification "deleting" event.
+     * Handle the notification "deleted" event.
      *
      * @param  \App\StatusHashtag  $hashtag
      * @return void
      */
-    public function deleting(StatusHashtag $hashtag)
+    public function deleted(StatusHashtag $hashtag)
     {
         StatusHashtagService::del($hashtag->hashtag_id, $hashtag->status_id);
         DB::table('hashtags')->where('id', $hashtag->hashtag_id)->decrement('cached_count');

app/Services/HashtagFollowService.php

@@ -11,11 +11,67 @@
 class HashtagFollowService
 {
 	const FOLLOW_KEY = 'pf:services:hashtag-follows:v1:';
+	const CACHE_KEY = 'pf:services:hfs:byHid:';
+	const CACHE_WARMED = 'pf:services:hfs:wc:byHid';
 
 	public static function getPidByHid($hid)
 	{
-		return Cache::remember(self::FOLLOW_KEY . $hid, 86400, function() use($hid) {
-			return HashtagFollow::whereHashtagId($hid)->pluck('profile_id')->toArray();
-		});
+		if(!self::isWarm($hid)) {
+			return self::warmCache($hid);
+		}
+		return self::get($hid);
+	}
+
+	public static function unfollow($hid, $pid)
+	{
+		$list = self::getPidByHid($hid);
+		if($list && count($list)) {
+			$list = array_values(array_diff($list, [$pid]));
+			Cache::put(self::FOLLOW_KEY . $hid, $list, 86400);
+		}
+		return;
+	}
+
+	public static function add($hid, $pid)
+	{
+		return Redis::zadd(self::CACHE_KEY . $hid, $pid, $pid);
+	}
+
+	public static function rem($hid, $pid)
+	{
+		return Redis::zrem(self::CACHE_KEY . $hid, $pid);
+	}
+
+	public static function get($hid)
+	{
+		return Redis::zrange(self::CACHE_KEY . $hid, 0, -1);
+	}
+
+	public static function count($hid)
+	{
+		return Redis::zcard(self::CACHE_KEY . $hid);
+	}
+
+	public static function warmCache($hid)
+	{
+		foreach(HashtagFollow::whereHashtagId($hid)->lazyById(20, 'id') as $h) {
+			if($h) {
+				self::add($h->hashtag_id, $h->profile_id);
+			}
+		}
+
+		self::setWarm($hid);
+
+		return self::get($hid);
+	}
+
+	public static function isWarm($hid)
+	{
+		return Redis::zcount($hid, 0, -1) ?? Redis::zscore(self::CACHE_WARMED, $hid) != null;
+	}
+
+	public static function setWarm($hid)
+	{
+		return Redis::zadd(self::CACHE_WARMED, $hid, $hid);
 	}
 }